15 Commits

Author SHA1 Message Date
f11c56e890 fix: make direct campaign pages scrollable 2026-07-28 22:50:11 +02:00
8d0a2608ed Declare campaign route View surfaces 2026-07-28 21:04:55 +02:00
89ae14c032 Integrate campaign delivery with hierarchical mail profiles 2026-07-28 19:33:12 +02:00
38af25ee88 docs: update GovOPlaN repository links 2026-07-27 15:46:51 +02:00
4774a8025c fix(campaign): roll back rejected immediate queues 2026-07-22 20:30:00 +02:00
689dc1fd6b fix(campaign): honor acceptance temp selection 2026-07-22 20:20:31 +02:00
90677348ff docs(campaign): define broker redelivery evidence 2026-07-22 20:19:51 +02:00
06786e86ef test(campaign): prove Celery broker redelivery 2026-07-22 20:19:29 +02:00
6fda123fc3 docs(campaign): distinguish worker process evidence 2026-07-22 18:26:42 +02:00
2fa91bb943 test(campaign): complete delivery fault drills 2026-07-22 18:26:28 +02:00
e3cc476508 fix(campaign): initialize delivery worker runtime 2026-07-22 18:26:11 +02:00
01ef541917 docs(campaign): define target-like mail evidence 2026-07-22 16:49:57 +02:00
d567257311 test(campaign): prove GreenMail delivery journey 2026-07-22 16:49:57 +02:00
3075ef7f5b docs(campaign): document aggregate report baseline 2026-07-22 15:20:04 +02:00
1ab4e91ffd test(campaign): add simple announcement acceptance fixture 2026-07-22 15:18:52 +02:00
42 changed files with 5030 additions and 135 deletions

View File

@@ -94,7 +94,7 @@ Platform RBAC and governance rules are documented in `govoplan-core/docs/`.
- [Campaign handbook](docs/CAMPAIGN_HANDBOOK.md) provides the adaptive user, process, governance, technical, and operations perspectives.
- [Campaign delivery runbook](docs/CAMPAIGN_DELIVERY_RUNBOOK.md) covers queueing, local vs Celery operation, retries, reconciliation, reports, and the live SMTP/IMAP test checklist.
- Immediate delivery is bounded to 25 exact eligible recipient jobs by default. Deployments may set `GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS` (0500), and tenants may narrow that ceiling through `campaign_delivery_policy.synchronous_send_max_recipients` in tenant settings.
- Report-email preview uses the selected version's stored v5 Mail-profile evidence. Live report email fails closed until [govoplan-mail#17](https://git.add-ideas.de/add-ideas/govoplan-mail/issues/17) provides a durable, idempotent Mail-owned outbox and transport-attempt ledger; per-job CSV is off by default and requires `campaigns:recipient:export` when requested.
- Report-email preview uses the selected version's stored v5 Mail-profile evidence. Live report email fails closed until [govoplan-mail#17](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17) provides a durable, idempotent Mail-owned outbox and transport-attempt ledger; per-job CSV is off by default and requires `campaigns:recipient:export` when requested.
- [Campaign/Mail profile boundary](docs/MAIL_PROFILE_BOUNDARY.md) defines profile-only delivery, runtime resolution, execution evidence, and the fail-closed legacy migration path.
- [Recipient import guide](docs/RECIPIENT_IMPORT_GUIDE.md) covers user/admin workflows, mapping profiles, validation, and import evidence.
- [Recipient and address boundary](docs/RECIPIENT_ADDRESS_BOUNDARY.md) defines the split between campaign-local recipients and future reusable address management.

View File

@@ -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

View File

@@ -44,6 +44,111 @@ If the smoke is started immediately after `docker compose up -d`, GreenMail may
bind the SMTP/IMAP ports before the services are fully ready. The smoke retries
login and folder setup for `GOVOPLAN_MAIL_TEST_READY_TIMEOUT_SECONDS`.
## Campaign Acceptance
The transport smoke proves the Mail adapters. The Campaign acceptance runner
proves the public composition: it creates an isolated temporary Core database,
creates a Mail-owned encrypted profile through the API, materializes the
credential-free [`greenmail-delivery`](../../examples/greenmail-delivery/campaign.json)
fixture with only that profile reference, validates/builds it, sends the exact
generated EML through Campaign once, appends it once, and cross-checks Campaign
report/audit state with one unique-subject message in the GreenMail INBOX and
Sent folders. Provider mailbox verification is not a byte-for-byte comparison
after provider-side header or storage transformations.
```bash
cd /mnt/DATA/git/govoplan-campaign/dev/mail-testbed
set -a
. ./.env
set +a
/mnt/DATA/git/govoplan/.venv/bin/python run_campaign_acceptance.py \
--evidence /tmp/govoplan-campaign-greenmail-evidence.json
```
The default run also uses controlled loopback protocol endpoints to prove that
an SMTP connection loss before transmission is temporary, an explicit SMTP
authentication rejection is permanent, a final `451` response after DATA is
temporary, one accepted and one refused RCPT command is retained as partial
envelope acceptance, a connection loss after complete DATA is frozen as
`outcome_unknown`, and an IMAP authentication rejection after SMTP acceptance
leaves the send accepted while the append fails. A
second ordinary send must be rejected before another provider effect.
The worker drill queues one job, starts the registered
`govoplan.campaigns.send_email` task body in a dedicated OS process, waits until
the controlled endpoint has received complete DATA, terminates that process,
and starts the same task body in a fresh process. It proves the durable
`sending`/unfinished-attempt boundary is recovered as `outcome_unknown`
without a second SMTP connection or DATA transaction. This is a real process
and task-boundary interruption, but it does not start a Celery daemon, Redis
broker, or broker redelivery; `celery_broker_redelivery` therefore remains
`false` in the evidence.
The bounded JSON contains no endpoint, account, address, credential, profile,
campaign, version, or job identifiers. It records module versions, the fixture
hash, normalized classifications/counts, the Mail-profile boundary, required
audit actions, provider mailbox increments, and coverage flags. Runtime version
declarations identify the exercised composition; they do not claim that the
sources are clean, tagged, signed, or release-provenanced. The evidence names
them `declared_module_versions` and keeps `source_artifact_provenance` false;
exact commit/artifact provenance belongs to the package and release gate.
This runner is restricted to literal loopback IP addresses and the synchronous
Campaign delivery mode. Hostnames such as `localhost` and every non-loopback
address fail before profile creation, avoiding a DNS change between validation
and connection. It is local target-like evidence, not approval of an
institution's SMTP/IMAP service. The controlled post-DATA, temporary-response,
partial-refusal, and task-process interruption drills are local effect-level
proof, not proof of a target provider's behavior or Redis/Celery broker
redelivery. Use `--success-only` only when testing the success journey without
the local failure endpoints.
## Redis/Celery Redelivery Acceptance
Run the maintained broker/worker-loss acceptance separately from the GreenMail
journey:
```bash
cd /mnt/DATA/git/govoplan-campaign/dev/mail-testbed
set -a
. ./.env
set +a
/mnt/DATA/git/govoplan/.venv/bin/python run_celery_redelivery_acceptance.py \
--evidence /tmp/govoplan-campaign-celery-redelivery-evidence.json
```
The runner creates a unique Compose project, starts only its loopback-bound,
AOF-enabled Redis service, creates an isolated temporary GovOPlaN database,
and starts a real Celery worker subscribed to `send_email`. A controlled SMTP
server holds the transaction after complete DATA and before the final response.
The runner kills that solo worker with the task still unacknowledged and starts
a replacement worker. After the configured Redis visibility timeout, the same Celery task identity must be redelivered.
The replacement must turn the durable
unfinished attempt into `outcome_unknown`, acknowledge the task, drain the
broker queue/unacked records, and leave the SMTP endpoint at exactly one
connection and one DATA transaction.
Only bounded counts, classifications, and booleans are retained. Worker logs,
task IDs, database identifiers, endpoints, credentials, and raw diagnostics are
kept in the temporary runtime and deleted. The evidence proves the local Redis
transport, real Celery process boundary, runner-supervised replacement, and
Campaign's duplicate-effect guard. It deliberately keeps production daemon supervision,
target-provider behavior, and source-artifact provenance false.
It does not claim that systemd, Kubernetes, another container orchestrator, or
an institution's Redis/SMTP deployment behaves identically.
The default run requires Docker CLI/Compose/daemon access and permission to
pull `redis:7-alpine`; the Celery workers execute from the current Python
environment. The isolated Compose project and volume are removed on exit.
`GOVOPLAN_CAMPAIGN_TEST_REDIS_VISIBILITY_TIMEOUT_SECONDS` defaults to three
seconds only to make this destructive local drill finish promptly; it is not a
production recommendation.
Starting the maintained test bed requires a working Docker CLI, Compose plugin,
daemon/socket access, and permission to pull `greenmail/standalone:2.1.9`. The
Campaign runner needs only the already-running loopback endpoints; it neither
starts Docker nor claims that it did.
## Use With A Campaign
Use the same settings in a campaign mail profile:

View File

@@ -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:

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,856 @@
#!/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 _create_runtime_root() -> Path:
"""Create the isolated runtime under the platform-selected temp root."""
return Path(tempfile.mkdtemp(prefix="govoplan-campaign-celery-redelivery-"))
def _bootstrap_and_run(
*,
settings: TestbedSettings,
fixture_path: Path,
redis_url: str,
visibility_timeout_seconds: int,
) -> dict[str, Any]:
runtime_root = _create_runtime_root()
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())

View File

@@ -104,6 +104,36 @@ bed where possible:
- IMAP append failure after SMTP acceptance.
- Worker restart with queued, claimed, and sending jobs.
For the maintained loopback baseline, run
`dev/mail-testbed/run_campaign_acceptance.py`. It proves the public Campaign
path for SMTP acceptance, IMAP append, repeat-send blocking, an SMTP connection
failure before transmission, an explicit SMTP authentication rejection, an
explicit temporary `451` response after DATA, partial RCPT refusal, a
connection loss after complete DATA, and an IMAP authentication rejection
after SMTP acceptance. Its evidence is an allowlisted classification/count
projection; raw provider diagnostics and transport/account identifiers are
deliberately excluded.
The runner also terminates a dedicated OS process executing the registered
Campaign send task after complete DATA, then invokes the task in a fresh
process. The unfinished durable attempt must become `outcome_unknown` and the
endpoint must observe no second connection or DATA transaction. This covers
the worker task/process boundary but not a broker or daemon.
Run `dev/mail-testbed/run_celery_redelivery_acceptance.py` for the maintained
Redis/Celery delivery and broker redelivery boundary. It starts an isolated
Redis Compose service and real Celery workers, kills the first solo worker after complete DATA while the
late-ack task is unacknowledged, and requires the same task identity to reach a
replacement worker after Redis visibility recovery. Passing evidence also
requires durable `outcome_unknown`, an empty broker queue/unacked set, and
exactly one SMTP connection and DATA transaction. Raw worker logs and task,
database, endpoint, and credential identifiers are never retained.
That second runner proves local runner-supervised process replacement, not the
production process manager. Repeat the worker-loss drill under the selected
systemd, container, Kubernetes, or other production supervisor and the target
Redis/SMTP infrastructure before deployment approval.
## Reporting Checks
- Partial delivery must show accepted, failed, and unknown counts separately.

View File

@@ -144,10 +144,14 @@ paths, storage keys, worker claim tokens, and raw provider diagnostics require
the dedicated diagnostic permission and must not leak through ordinary campaign,
version, job, or report responses.
The current Campaign Report Web UI additionally requires recipient-read access
and does not yet hide every action control that the actor lacks. The server
still authorizes each action, but an aggregate-only reader UI remains open
work; do not promise that experience from `campaigns:report:read` alone.
Campaign now provides a separate aggregate **Reports** surface for readers with
`campaigns:report:read` and access to the campaign. It loads only the safe
aggregate projections, applies small-cell suppression, and offers no recipient
rows, drill-down, filtering, export, or delivery actions. The recipient-aware
**Campaign Report** still requires recipient-read access and does not yet hide
every action control that the actor lacks. The server authorizes each action,
but permission-aware action visibility on that detailed surface remains open
work; do not confuse it with the aggregate reader experience.
### Deliver and resolve outcomes
@@ -465,7 +469,7 @@ the current baseline:
- the final audited **test / single send / single resend** semantics;
- reusable SMTP batch sessions and their measured throughput benefit;
- durable, idempotent Campaign report delivery through a Mail-owned outbox
([`govoplan-mail#17`](https://git.add-ideas.de/add-ideas/govoplan-mail/issues/17));
([`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17));
- a fully packaged one-command Campaign reference composition with production
policy presets and target-provider certification;
- function-bound Postbox delivery (stage 2 of the reference program);

View File

@@ -10,7 +10,11 @@ scenario catalogue lives in `examples/README.md`; committed fixture files should
be added under `examples/` only when they validate against the current campaign
schema and are safe to run in non-production environments.
- simple announcement with one active recipient and no attachments
- [`simple-announcement`](../examples/simple-announcement/campaign.json), a
credential-free campaign with one active recipient and no attachments; its
automated acceptance check physically blocks Mail and Files imports, denies
network connections, and validates/builds from an unrelated temporary
workspace
- multi-recipient message with To, CC, BCC, Reply-To, bounce, and disposition
notification fields
- campaign with global attachments and recipient-specific attachment rules
@@ -22,7 +26,9 @@ schema and are safe to run in non-production environments.
- campaign with blocked recipients or attachment errors that must not be sent
- mock delivery campaign that captures SMTP and IMAP append messages in the mail
development mailbox
- real non-production delivery campaign against the GreenMail test bed
- [`greenmail-delivery`](../examples/greenmail-delivery/campaign.json), a
credential-free real-delivery Campaign materialized with a temporary
Mail-owned profile by the loopback acceptance runner
## Fixture Rules
@@ -41,12 +47,31 @@ Before tagging a campaign release:
- Review `examples/README.md` and update the scenario catalogue when a release
adds or removes delivery behavior.
- Run `python -m unittest discover -s tests -p 'test_example_campaigns.py'` and
retain its isolated validate/build result as release evidence.
- Run core module permutation tests with campaign installed both with and
without files/mail.
- Validate and build each maintained example campaign.
- Run the mock delivery example when the mail development mailbox capability is
enabled.
- Run the GreenMail SMTP/IMAP smoke for a non-production real delivery path.
- Run `dev/mail-testbed/run_campaign_acceptance.py` and retain its bounded JSON
projection. It must show one SMTP acceptance, one IMAP append, no duplicate
effect from a repeated ordinary send, matching Campaign report/audit state,
and no resolved transport material in Campaign JSON or its execution
snapshot. Its controlled endpoint evidence must also show explicit SMTP 451,
partial RCPT refusal, post-DATA ambiguity, and task-process interruption
classifications without retaining addresses or provider diagnostics.
- Treat the task-process restart proof separately from the still-open
Redis/Celery broker redelivery and daemon-supervision check; the coverage
projection must keep `celery_broker_redelivery` false.
- Run `dev/mail-testbed/run_celery_redelivery_acceptance.py` as a separate
destructive worker-loss check. Retain its bounded evidence only when the same
broker task is observed at both workers, the durable state is
`outcome_unknown`, broker queue/unacked counts are zero, and the controlled
SMTP endpoint observed one connection and one DATA transaction. This closes
local Redis/Celery redelivery coverage, while production supervisor and
target-provider coverage remain false until separately tested.
- Confirm reusable mail profile selection is revalidated after campaign owner
transfer.
- Confirm every inline SMTP/IMAP field is rejected on import/write, omitted

View File

@@ -49,7 +49,7 @@ Non-dry Campaign report email currently fails closed. It must not bypass the
durable job/effect model through a direct SMTP call. Re-enabling it requires the
Mail-owned idempotent outbox, attempt, unknown-outcome, and reconciliation path
tracked in
[`govoplan-mail#17`](https://git.add-ideas.de/add-ideas/govoplan-mail/issues/17).
[`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17).
Report generation and dry-run validation remain separate from an external
effect; recipient-level exports require recipient-export authorization.

View File

@@ -9,7 +9,7 @@ campaign schema and do not require production data.
| Scenario | Required Modules | Release Check |
| --- | --- | --- |
| `simple-announcement` | core, access, campaigns | Validate and build one active recipient without attachments. |
| [`simple-announcement`](simple-announcement/campaign.json) | core, access, campaigns | Validate and build one active recipient without attachments while Mail and Files are absent. |
| `addressing-matrix` | core, access, campaigns | Exercise To, CC, BCC, Reply-To, bounce, and disposition-notification fields. |
| `global-attachment` | core, access, campaigns; optional files | Build one deterministic attachment and verify evidence. |
| `recipient-attachment-rules` | core, access, campaigns; optional files | Match recipient-specific attachment rules and verify per-recipient evidence. |
@@ -19,7 +19,7 @@ campaign schema and do not require production data.
| `warnings-review` | core, access, campaigns | Require explicit review before queueing jobs with warnings. |
| `blocked-send` | core, access, campaigns | Confirm blocked recipients or missing attachments cannot be queued. |
| `mock-delivery` | core, access, campaigns, mail with dev capability | Capture messages in the development mailbox. |
| `greenmail-delivery` | core, access, campaigns, mail | Send no-attachment, normal attachment, and ZIP attachment variants through `dev/mail-testbed`. |
| [`greenmail-delivery`](greenmail-delivery/campaign.json) | core, access, audit, campaigns, mail | Run a credential-free Campaign through a Mail-owned profile, GreenMail SMTP/IMAP, report/audit checks, repeat-send protection, and bounded failure drills. |
## Fixture Rules
@@ -37,9 +37,14 @@ campaign schema and do not require production data.
Before a release tag:
1. Run module permutation startup checks from core.
2. Validate every committed example fixture against the current campaign schema.
3. Build exact messages for each fixture.
4. Run the mock-delivery example when the dev mailbox capability is enabled.
5. Run `dev/mail-testbed/run_transport_smoke.py`.
6. Execute the delivery checklist in
2. Run `python -m unittest discover -s tests -p 'test_example_campaigns.py'`
from this repository. The acceptance test copies each maintained fixture to
an unrelated temporary workspace before using Campaign's public loader,
validator, and message builder.
3. Validate every committed example fixture against the current campaign schema.
4. Build exact messages for each fixture.
5. Run the mock-delivery example when the dev mailbox capability is enabled.
6. Run `dev/mail-testbed/run_transport_smoke.py` for low-level transport and attachment variants.
7. Run `dev/mail-testbed/run_campaign_acceptance.py` for the Campaign journey and bounded evidence.
8. Execute the delivery checklist in
`docs/EXAMPLE_CAMPAIGNS_AND_RELEASE_CHECKLIST.md`.

View File

@@ -0,0 +1,88 @@
{
"version": "1.0",
"campaign": {
"id": "greenmail-delivery",
"name": "GreenMail delivery acceptance",
"description": "Credential-free Campaign fixture for the local SMTP/IMAP acceptance test bed.",
"mode": "test"
},
"fields": [
{
"name": "display_name",
"type": "string",
"label": "Display name",
"required": true
},
{
"name": "acceptance_run",
"type": "string",
"label": "Acceptance run",
"required": true
}
],
"server": {
"mail_profile_id": "00000000-0000-4000-8000-000000000001"
},
"recipients": {
"from": [
{
"email": "campaign-test@govoplan.test",
"name": "GovOPlaN acceptance",
"type": "to"
}
],
"allow_individual_to": true
},
"template": {
"subject": "[GovOPlaN acceptance ${acceptance_run}] Campaign delivery",
"text": "Hello ${display_name},\n\nThis is an isolated GovOPlaN Campaign SMTP/IMAP acceptance message.\n",
"body_mode": "text"
},
"attachments": {
"base_path": ".",
"send_without_attachments_behavior": "continue",
"global": []
},
"entries": {
"inline": [
{
"id": "greenmail-recipient",
"to": [
{
"email": "campaign-test@govoplan.test",
"name": "GreenMail recipient",
"type": "to"
}
],
"fields": {
"display_name": "GreenMail recipient",
"acceptance_run": "fixture"
}
}
]
},
"validation_policy": {
"missing_email": "block",
"template_error": "block"
},
"delivery": {
"rate_limit": {
"messages_per_minute": 60
},
"retry": {
"max_attempts": 3,
"backoff_seconds": [
1,
5,
30
]
},
"imap_append_sent": {
"enabled": true,
"folder": "Sent"
}
},
"status_tracking": {
"enabled": true
}
}

View File

@@ -0,0 +1,22 @@
{
"scenario": "greenmail-delivery",
"campaign_file": "campaign.json",
"required_modules": [
"core",
"access",
"audit",
"campaigns",
"mail"
],
"required_capabilities": [
"mail.campaign_delivery"
],
"transport": "local GreenMail SMTP/IMAP test bed",
"credentials": "local environment only; never copied into Campaign JSON or evidence",
"expected": {
"entries_count": 1,
"built_count": 1,
"smtp_accepted_count": 1,
"imap_appended_count": 1
}
}

View File

@@ -0,0 +1,54 @@
{
"version": "1.0",
"campaign": {
"id": "simple-announcement",
"name": "Simple announcement",
"description": "Credential-free release fixture for Campaign validation and message building.",
"mode": "test"
},
"fields": [
{
"name": "display_name",
"type": "string",
"label": "Display name",
"required": true
}
],
"recipients": {
"from": [
{
"email": "announcements@example.test",
"name": "GovOPlaN Example",
"type": "to"
}
],
"allow_individual_to": true
},
"template": {
"subject": "Planned service maintenance for ${display_name}",
"text": "Hello ${display_name},\n\nThe example service will be unavailable during the announced maintenance window.\n\nThis message was built locally and was not sent.\n",
"body_mode": "text"
},
"attachments": {
"base_path": ".",
"send_without_attachments_behavior": "continue",
"global": []
},
"entries": {
"inline": [
{
"id": "example-recipient",
"to": [
{
"email": "recipient@example.test",
"name": "Example Recipient",
"type": "to"
}
],
"fields": {
"display_name": "Example Recipient"
}
}
]
}
}

View File

@@ -0,0 +1,23 @@
{
"schema_version": 1,
"id": "simple-announcement",
"campaign_file": "campaign.json",
"required_modules": [
"core",
"access",
"campaigns"
],
"absent_optional_modules": [
"files",
"mail"
],
"external_effects": "forbidden",
"expected": {
"campaign_id": "simple-announcement",
"entries_count": 1,
"built_count": 1,
"queueable_count": 1,
"attachment_count": 0,
"subject": "Planned service maintenance for Example Recipient"
}
}

View File

@@ -1,6 +1,6 @@
{
"name": "@govoplan/campaign-webui",
"version": "0.1.10",
"version": "0.1.11",
"private": true,
"type": "module",
"main": "webui/src/index.ts",

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-campaign"
version = "0.1.10"
version = "0.1.11"
description = "GovOPlaN campaigns module with backend and WebUI integration."
readme = "README.md"
requires-python = ">=3.12"

View File

@@ -4,7 +4,15 @@ import copy
from typing import Any
CAMPAIGN_MAIL_SERVER_KEYS = frozenset({"mail_profile_id"})
CAMPAIGN_MAIL_SERVER_KEYS = frozenset(
{
"mail_profile_id",
"smtp_server_id",
"smtp_credential_id",
"imap_server_id",
"imap_credential_id",
}
)
CAMPAIGN_CLIENT_EDITOR_STATE_KEYS = frozenset({"created_from", "field_overrides", "opt_ins"})
CAMPAIGN_OPT_IN_KEYS = frozenset(
{"campaign_address_suggestions", "remember_used_addresses", "inline_guidance"}
@@ -24,8 +32,8 @@ class CampaignMailProfileBoundaryError(ValueError):
"""Raised when campaign JSON owns mail transport configuration.
SMTP/IMAP endpoints and credentials are Mail-module data. Campaign JSON
may select one Mail-owned profile, but it must never copy or override that
profile's transport configuration.
may select Mail-owned profile, server, and credential identifiers, but it
must never copy or override transport configuration.
"""
@@ -218,16 +226,53 @@ def campaign_mail_profile_id(raw_json: dict[str, Any] | None) -> str | None:
return normalized or None
def campaign_mail_resource_ids(
raw_json: dict[str, Any] | None,
) -> dict[str, str | None]:
server = raw_json.get("server") if isinstance(raw_json, dict) else None
if not isinstance(server, dict):
return {
"mail_profile_id": None,
"smtp_server_id": None,
"smtp_credential_id": None,
"imap_server_id": None,
"imap_credential_id": None,
}
return {
key: (
value.strip()
if isinstance((value := server.get(key)), str) and value.strip()
else None
)
for key in CAMPAIGN_MAIL_SERVER_KEYS
}
def campaign_mail_profile_boundary_violations(raw_json: dict[str, Any] | None) -> tuple[str, ...]:
server = raw_json.get("server") if isinstance(raw_json, dict) else None
if not isinstance(server, dict):
return ()
violations = [f"/server/{key}" for key in sorted(server) if key not in CAMPAIGN_MAIL_SERVER_KEYS]
if "mail_profile_id" in server:
profile_id = server["mail_profile_id"]
if not isinstance(profile_id, str) or not profile_id.strip():
for key in CAMPAIGN_MAIL_SERVER_KEYS:
if key not in server:
continue
value = server[key]
if not isinstance(value, str) or not value.strip():
violations.append(f"/server/{key}")
references = campaign_mail_resource_ids(raw_json)
if references["mail_profile_id"] is None and any(
references[key]
for key in references
if key != "mail_profile_id"
):
violations.append("/server/mail_profile_id")
for protocol in ("smtp", "imap"):
if (
references[f"{protocol}_credential_id"]
and not references[f"{protocol}_server_id"]
):
violations.append(f"/server/{protocol}_server_id")
return tuple(violations)
@@ -240,9 +285,9 @@ def assert_campaign_uses_mail_profile_reference(
if violations:
fields = ", ".join(violations)
raise CampaignMailProfileBoundaryError(
"Campaign JSON may only reference a Mail-module profile through "
f"server.mail_profile_id; remove campaign-local SMTP/IMAP settings ({fields}), "
"select an authorized Mail profile, and save a new campaign version."
"Campaign JSON may only reference Mail-owned profiles, servers, and credentials; "
f"remove campaign-local SMTP/IMAP settings or invalid references ({fields}), "
"select authorized Mail resources, and save a new campaign version."
)
if require_profile and campaign_mail_profile_id(raw_json) is None:
raise CampaignMailProfileBoundaryError(
@@ -254,5 +299,8 @@ def assert_campaign_uses_mail_profile_reference(
def public_campaign_mail_server(raw_json: dict[str, Any] | None) -> dict[str, str]:
"""Return the complete public/persisted Campaign-to-Mail contract."""
profile_id = campaign_mail_profile_id(raw_json)
return {"mail_profile_id": profile_id} if profile_id else {}
return {
key: value
for key, value in campaign_mail_resource_ids(raw_json).items()
if value
}

View File

@@ -114,6 +114,10 @@ class MailProfileCapabilities(StrictModel):
class ServerConfig(StrictModel):
mail_profile_id: str | None = None
smtp_server_id: str | None = None
smtp_credential_id: str | None = None
imap_server_id: str | None = None
imap_credential_id: str | None = None
profile_capabilities: MailProfileCapabilities = Field(default_factory=MailProfileCapabilities)

View File

@@ -279,6 +279,12 @@ class CampaignDeliveryTaskService(CampaignDeliveryTaskProvider):
def delivery_tasks_capability(context: object) -> CampaignDeliveryTaskService:
from govoplan_campaign.backend.runtime import configure_runtime
configure_runtime(
registry=getattr(context, "registry", None),
settings=getattr(context, "settings", None),
)
return CampaignDeliveryTaskService()

View File

@@ -16,6 +16,7 @@ from govoplan_core.core.modules import (
DocumentationLink,
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
@@ -156,7 +157,7 @@ def _campaigns_router(context: ModuleContext):
manifest = ModuleManifest(
id="campaigns",
name="Campaigns",
version="0.1.10",
version="0.1.11",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
optional_dependencies=("files", "mail", "notifications", "addresses"),
provides_interfaces=(
@@ -216,6 +217,39 @@ manifest = ModuleManifest(
frontend=FrontendModule(
module_id="campaigns",
package_name="@govoplan/campaign-webui",
routes=(
FrontendRoute(
path="/campaigns",
component="CampaignListPage",
required_any=("campaigns:campaign:read",),
order=20,
),
FrontendRoute(
path="/campaigns/:campaignId/*",
component="CampaignWorkspace",
required_any=("campaigns:campaign:read",),
order=21,
),
FrontendRoute(
path="/operator",
component="OperatorQueuePage",
required_all=("campaigns:campaign:read",),
required_any=(
"campaigns:campaign:queue",
"campaigns:campaign:retry",
"campaigns:campaign:reconcile",
"campaigns:campaign:control",
),
order=30,
),
FrontendRoute(
path="/reports",
component="AggregateReportsPage",
required_any=("campaigns:report:read",),
order=70,
),
FrontendRoute(path="/templates", component="TemplatesPage", order=90),
),
nav_items=(
NavItem(path="/campaigns", label="Campaigns", icon="campaign", required_any=("campaigns:campaign:read",), order=20),
NavItem(

View File

@@ -30,11 +30,12 @@ from govoplan_campaign.backend.campaign.loader import load_campaign_json, valida
from govoplan_campaign.backend.campaign.mail_profile_boundary import (
assert_campaign_uses_mail_profile_reference,
campaign_mail_profile_id,
campaign_mail_resource_ids,
)
from govoplan_campaign.backend.campaign.validation import validate_campaign_config
from govoplan_campaign.backend.messages.builder import build_campaign_messages
from govoplan_campaign.backend.messages.models import MessageDraft
from govoplan_campaign.backend.sending.execution import create_execution_snapshot, profile_transport_revisions
from govoplan_campaign.backend.sending.execution import create_execution_snapshot, profile_delivery_summary
from govoplan_campaign.backend.campaign.models import CampaignConfig, SendStatus
from govoplan_campaign.backend.integrations import files_integration, mail_integration
from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths
@@ -73,6 +74,7 @@ def load_campaign_config_from_json(
materialized = copy.deepcopy(raw_json)
profile_id = campaign_mail_profile_id(raw_json)
if profile_id:
references = campaign_mail_resource_ids(raw_json)
summary = mail_integration().campaign_profile_delivery_summary(
session,
tenant_id=tenant_id,
@@ -80,6 +82,10 @@ def load_campaign_config_from_json(
profile_id=profile_id,
owner_user_id=owner_user_id,
owner_group_id=owner_group_id,
smtp_server_id=references["smtp_server_id"],
smtp_credential_id=references["smtp_credential_id"],
imap_server_id=references["imap_server_id"],
imap_credential_id=references["imap_credential_id"],
)
materialized.setdefault("server", {})["profile_capabilities"] = {
"smtp_available": bool(summary.get("smtp_available")),
@@ -513,14 +519,18 @@ def build_campaign_version(
profile_id = campaign_mail_profile_id(version.raw_json if isinstance(version.raw_json, dict) else {})
if profile_id is None:
raise CampaignPersistenceError("Select an authorized Mail profile before building campaign messages")
revisions = profile_transport_revisions(session, version)
if not revisions["smtp"]:
profile_summary = profile_delivery_summary(session, version)
if not profile_summary.get("smtp_transport_revision"):
raise CampaignPersistenceError("The selected Mail profile has no SMTP transport revision")
execution_snapshot, execution_snapshot_hash = create_execution_snapshot(
version,
mail_profile_id=profile_id,
smtp_transport_revision=revisions["smtp"],
imap_transport_revision=revisions["imap"],
smtp_server_id=profile_summary.get("smtp_server_id"),
smtp_credential_id=profile_summary.get("smtp_credential_id"),
imap_server_id=profile_summary.get("imap_server_id"),
imap_credential_id=profile_summary.get("imap_credential_id"),
smtp_transport_revision=profile_summary["smtp_transport_revision"],
imap_transport_revision=profile_summary.get("imap_transport_revision"),
delivery=managed_config.delivery,
jobs=[job for job, _message in job_build_pairs],
build_summary=report_json,

View File

@@ -136,7 +136,10 @@ from govoplan_campaign.backend.integrations import (
)
from govoplan_campaign.backend.path_security import CampaignPathSecurityError, assert_server_safe_campaign_paths
from govoplan_campaign.backend.campaign.loader import load_campaign_json
from govoplan_campaign.backend.campaign.mail_profile_boundary import campaign_mail_profile_id
from govoplan_campaign.backend.campaign.mail_profile_boundary import (
CAMPAIGN_MAIL_SERVER_KEYS,
campaign_mail_profile_id,
)
from govoplan_campaign.backend.attachments.resolver import resolve_campaign_attachments
from govoplan_core.security.time import utc_now
from govoplan_campaign.backend.persistence.versions import (
@@ -581,7 +584,8 @@ def _clear_current_version_mail_profile_for_owner_transfer(session: Session, cam
)
next_server = dict(server)
next_server.pop("mail_profile_id", None)
for key in CAMPAIGN_MAIL_SERVER_KEYS:
next_server.pop(key, None)
next_server.pop("profile_id", None)
raw_json["server"] = next_server
@@ -3608,6 +3612,10 @@ def send_campaign_now_endpoint(
)
return SendCampaignNowResponse(result=response_result)
except SynchronousSendRejected as exc:
# A synchronous request stages queue state before the all-message
# preflight can run. Rejecting that preflight must not leave work
# eligible for a background worker when no provider effect occurred.
session.rollback()
audit_from_principal(
session,
principal,

View File

@@ -92,7 +92,27 @@
"mail_profile_id": {
"type": "string",
"minLength": 1,
"description": "Stable reference to an authorized profile owned by the Mail module. Campaign JSON never stores SMTP/IMAP settings or credentials."
"description": "Stable reference to an authorized server envelope owned by the Mail module. Campaign JSON never stores SMTP/IMAP settings or credentials."
},
"smtp_server_id": {
"type": "string",
"minLength": 1,
"description": "Optional explicit Mail-owned SMTP server selection."
},
"smtp_credential_id": {
"type": "string",
"minLength": 1,
"description": "Optional explicit core credential envelope bound to the selected SMTP server."
},
"imap_server_id": {
"type": "string",
"minLength": 1,
"description": "Optional explicit Mail-owned IMAP server selection."
},
"imap_credential_id": {
"type": "string",
"minLength": 1,
"description": "Optional explicit core credential envelope bound to the selected IMAP server."
}
},
"additionalProperties": false

View File

@@ -14,11 +14,12 @@ from govoplan_campaign.backend.campaign.mail_profile_boundary import (
CampaignMailProfileBoundaryError,
assert_campaign_uses_mail_profile_reference,
campaign_mail_profile_id,
campaign_mail_resource_ids,
)
from govoplan_campaign.backend.integrations import MailProfileError, files_integration, mail_integration
from govoplan_campaign.backend.path_security import CampaignPathSecurityError, assert_server_safe_campaign_paths
SNAPSHOT_VERSION = "5"
SNAPSHOT_VERSION = "6"
class ExecutionSnapshotError(RuntimeError):
@@ -41,6 +42,10 @@ class ExecutionSnapshot(BaseModel):
campaign_version_id: str
campaign_json_sha256: str
mail_profile_id: str
smtp_server_id: str | None = None
smtp_credential_id: str | None = None
imap_server_id: str | None = None
imap_credential_id: str | None = None
created_at: str
build_token: str | None = None
built_at: str | None = None
@@ -75,12 +80,17 @@ def profile_delivery_summary(session: Session, version: CampaignVersion) -> dict
campaign = session.get(Campaign, version.campaign_id)
if campaign is None:
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
references = campaign_mail_resource_ids(raw_json)
try:
return mail.campaign_profile_delivery_summary(
session,
tenant_id=campaign.tenant_id,
campaign_id=campaign.id,
profile_id=profile_id,
smtp_server_id=references["smtp_server_id"],
smtp_credential_id=references["smtp_credential_id"],
imap_server_id=references["imap_server_id"],
imap_credential_id=references["imap_credential_id"],
)
except MailProfileError as exc:
raise ExecutionSnapshotError(str(exc)) from exc
@@ -102,6 +112,19 @@ def _assert_snapshot_profile_matches_version(version: CampaignVersion, snapshot:
"The campaign's Mail profile reference differs from the built execution snapshot. "
"Revalidate and rebuild the campaign before delivery."
)
references = campaign_mail_resource_ids(raw_json)
for key in (
"smtp_server_id",
"smtp_credential_id",
"imap_server_id",
"imap_credential_id",
):
configured = references[key]
if configured and configured != getattr(snapshot, key):
raise ExecutionSnapshotError(
"The campaign's Mail server or credential selection differs from the built "
"execution snapshot. Revalidate and rebuild before delivery."
)
def _assert_version_mail_profile_boundary(raw_json: dict[str, Any]) -> None:
@@ -163,6 +186,10 @@ def create_execution_snapshot(
smtp_transport_revision: str,
imap_transport_revision: str | None,
delivery: DeliveryConfig,
smtp_server_id: str | None = None,
smtp_credential_id: str | None = None,
imap_server_id: str | None = None,
imap_credential_id: str | None = None,
jobs: Iterable[CampaignJob] = (),
build_summary: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], str]:
@@ -176,6 +203,10 @@ def create_execution_snapshot(
campaign_version_id=version.id,
campaign_json_sha256=_sha256(raw_json),
mail_profile_id=mail_profile_id,
smtp_server_id=smtp_server_id,
smtp_credential_id=smtp_credential_id,
imap_server_id=imap_server_id,
imap_credential_id=imap_credential_id,
build_token=str(summary.get("build_token") or "") or None,
built_at=str(summary.get("built_at") or "") or None,
job_count=len(job_list),
@@ -316,14 +347,18 @@ def ensure_execution_snapshot(
)
if not jobs:
raise ExecutionSnapshotError("Campaign version has no built jobs; rebuild it before delivery")
revisions = profile_transport_revisions(session, version)
if not revisions["smtp"]:
summary = profile_delivery_summary(session, version)
if not summary.get("smtp_transport_revision"):
raise ExecutionSnapshotError("The selected Mail profile has no SMTP transport revision")
payload, digest = create_execution_snapshot(
version,
mail_profile_id=profile_id,
smtp_transport_revision=revisions["smtp"],
imap_transport_revision=revisions["imap"],
smtp_server_id=summary.get("smtp_server_id"),
smtp_credential_id=summary.get("smtp_credential_id"),
imap_server_id=summary.get("imap_server_id"),
imap_credential_id=summary.get("imap_credential_id"),
smtp_transport_revision=summary["smtp_transport_revision"],
imap_transport_revision=summary.get("imap_transport_revision"),
delivery=config.delivery,
jobs=jobs,
build_summary=version.build_summary if isinstance(version.build_summary, dict) else {},

View File

@@ -622,6 +622,7 @@ def _persist_campaign_queue(
version: CampaignVersion,
queued: list[CampaignJob],
delivery_mode: str,
commit: bool = True,
) -> None:
if queued:
previous_status = campaign.status
@@ -640,6 +641,7 @@ def _persist_campaign_queue(
version_id=version.id,
)
session.add(campaign)
if commit:
session.commit()
@@ -661,6 +663,7 @@ def queue_campaign_jobs(
include_warnings: bool = True,
dry_run: bool = False,
delivery_mode: str | None = None,
commit_queue: bool = True,
) -> QueueCampaignResult:
"""Move queueable DB jobs to QUEUED and optionally enqueue Celery tasks."""
@@ -694,6 +697,7 @@ def queue_campaign_jobs(
version=version,
queued=queued,
delivery_mode=selected_delivery_mode,
commit=commit_queue,
)
enqueued_count = _enqueue_campaign_jobs(
queued,
@@ -766,6 +770,7 @@ def send_campaign_now(
enqueue_celery=False,
dry_run=dry_run,
delivery_mode=DELIVERY_MODE_SYNCHRONOUS,
commit_queue=False,
)
if dry_run:
return SendCampaignNowResult(
@@ -802,6 +807,11 @@ def send_campaign_now(
jobs=jobs,
policy=synchronous_policy,
)
# Queue state and its inbox notification become durable only after every
# message and the selected transport revision have passed preflight. This
# preserves late-ack recovery without leaving rejected work eligible for a
# background worker.
session.commit()
results: list[dict[str, Any]] = []
sent_count = 0
@@ -1894,6 +1904,8 @@ def _send_claimed_campaign_job(
envelope_recipients=context.envelope_recipients,
from_header=_from_header_from_job(job),
expected_smtp_transport_revision=context.snapshot.smtp_transport_revision or "",
smtp_server_id=context.snapshot.smtp_server_id,
smtp_credential_id=context.snapshot.smtp_credential_id,
)
if result.accepted_count <= 0:
raise SmtpSendError("SMTP did not accept any envelope recipients", temporary=False)
@@ -2402,6 +2414,10 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False)
folder=None if folder == "auto" else folder,
expected_smtp_transport_revision=snapshot.smtp_transport_revision or "",
expected_imap_transport_revision=snapshot.imap_transport_revision,
smtp_server_id=snapshot.smtp_server_id,
smtp_credential_id=snapshot.smtp_credential_id,
imap_server_id=snapshot.imap_server_id,
imap_credential_id=snapshot.imap_credential_id,
)
except (MailProfileError, ImapConfigurationError, ImapAppendError) as exc:
outcome_unknown = bool(getattr(exc, "outcome_unknown", False))

View File

@@ -0,0 +1,314 @@
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_runbook_keeps_local_redelivery_distinct_from_production_supervision() -> None:
testbed = (REPOSITORY_ROOT / "dev" / "mail-testbed" / "README.md").read_text(
encoding="utf-8"
)
runbook = (REPOSITORY_ROOT / "docs" / "CAMPAIGN_DELIVERY_RUNBOOK.md").read_text(
encoding="utf-8"
)
assert "run_celery_redelivery_acceptance.py" in testbed
assert "same Celery task identity must be redelivered" in testbed
assert "production daemon supervision" in testbed
assert "empty broker queue/unacked set" in runbook
assert "production process manager" in runbook
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_runtime_root_uses_platform_temp_selection() -> None:
with mock.patch(
"govoplan_campaign_celery_redelivery_acceptance.tempfile.mkdtemp",
return_value="/selected-temp/govoplan-campaign-celery-redelivery-test",
) as mkdtemp:
runtime_root = runner._create_runtime_root()
assert runtime_root == Path(
"/selected-temp/govoplan-campaign-celery-redelivery-test"
)
mkdtemp.assert_called_once_with(prefix="govoplan-campaign-celery-redelivery-")
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

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@@ -364,6 +365,18 @@ def test_aggregate_report_task_never_implies_recipient_detail_or_export_authorit
assert "export" in topic.metadata["verification"].lower()
def test_handbook_distinguishes_shipped_aggregate_reports_from_detailed_report_gaps() -> None:
handbook = " ".join(
(
Path(__file__).resolve().parents[1] / "docs" / "CAMPAIGN_HANDBOOK.md"
).read_text(encoding="utf-8").lower().split()
)
assert "aggregate-only reader ui remains open" not in handbook
assert "separate aggregate **reports** surface" in handbook
assert "permission-aware action visibility on that detailed surface remains open work" in handbook
def test_static_campaign_handbook_has_unique_ids_help_contexts_and_no_planned_resend_claim() -> None:
from govoplan_campaign.backend.manifest import get_manifest

View File

@@ -0,0 +1,207 @@
from __future__ import annotations
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
FIXTURE_ROOT = REPOSITORY_ROOT / "examples" / "simple-announcement"
_ISOLATED_ACCEPTANCE_PROGRAM = r"""
from __future__ import annotations
import hashlib
import importlib.abc
import json
import socket
import sys
from email import policy
from email.parser import BytesParser
from pathlib import Path
source_root = Path(sys.argv[1]).resolve()
fixture_root = Path(sys.argv[2]).resolve()
sys.path.insert(0, str(source_root))
class AbsentOptionalModuleFinder(importlib.abc.MetaPathFinder):
absent_roots = {"govoplan_files", "govoplan_mail"}
def find_spec(self, fullname, path=None, target=None):
if fullname.partition(".")[0] in self.absent_roots:
raise ModuleNotFoundError(
f"{fullname} is intentionally absent in the Campaign fixture check",
name=fullname,
)
return None
sys.meta_path.insert(0, AbsentOptionalModuleFinder())
def deny_network(*args, **kwargs):
raise AssertionError("the Campaign validate/build fixture must not open a network connection")
class NoNetworkSocket(socket.socket):
def connect(self, address):
deny_network(address)
def connect_ex(self, address):
deny_network(address)
socket.create_connection = deny_network
socket.socket = NoNetworkSocket
from govoplan_campaign.backend.campaign import ( # noqa: E402
load_campaign_config,
load_campaign_json,
validate_campaign_config,
)
from govoplan_campaign.backend.messages import build_campaign_messages # noqa: E402
metadata = json.loads((fixture_root / "fixture.json").read_text(encoding="utf-8"))
assert metadata["required_modules"] == ["core", "access", "campaigns"]
assert metadata["absent_optional_modules"] == ["files", "mail"]
assert metadata["external_effects"] == "forbidden"
campaign_file = fixture_root / metadata["campaign_file"]
raw_campaign = load_campaign_json(campaign_file)
def iter_keys(value):
if isinstance(value, dict):
for key, nested in value.items():
yield key.casefold()
yield from iter_keys(nested)
elif isinstance(value, list):
for nested in value:
yield from iter_keys(nested)
forbidden_key_fragments = ("credential", "imap", "mail_profile", "password", "secret", "smtp")
assert not [
key
for key in iter_keys(raw_campaign)
if any(fragment in key for fragment in forbidden_key_fragments)
]
config = load_campaign_config(campaign_file)
assert config.server.mail_profile_id is None
assert not config.attachments.global_
validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True)
assert validation.ok
assert validation.error_count == 0
assert validation.warning_count == 0
assert validation.entries_count == metadata["expected"]["entries_count"]
def build(output_name):
return build_campaign_messages(
config,
campaign_file=campaign_file,
output_dir=fixture_root.parent / output_name,
write_eml=True,
)
first = build("build-first")
second = build("build-second")
expected = metadata["expected"]
for result in (first, second):
assert result.report.campaign_id == expected["campaign_id"]
assert result.report.entries_count == expected["entries_count"]
assert result.report.built_count == expected["built_count"]
assert result.report.build_failed_count == 0
assert result.report.queueable_count == expected["queueable_count"]
assert len(result.built_messages) == 1
built = result.built_messages[0]
assert built.mime is not None
assert built.draft.subject == expected["subject"]
assert built.draft.validation_status.value == "ready"
assert built.draft.send_status.value == "draft"
assert built.draft.imap_status.value == "not_requested"
assert built.draft.attachment_count == expected["attachment_count"]
assert not built.draft.attachments
assert not built.draft.issues
assert built.draft.from_ is not None
assert built.draft.from_.email == "announcements@example.test"
assert [address.email for address in built.draft.to] == ["recipient@example.test"]
assert built.mime["Subject"] == expected["subject"]
assert "Hello Example Recipient" in built.mime.get_content()
assert list(built.mime.iter_attachments()) == []
def normalized_eml(path_value):
message = BytesParser(policy=policy.default).parsebytes(Path(path_value).read_bytes())
del message["Date"]
del message["Message-ID"]
return message.as_bytes(policy=policy.default)
first_eml = normalized_eml(first.report.messages[0].eml_path)
second_eml = normalized_eml(second.report.messages[0].eml_path)
assert first_eml == second_eml
assert not any(
name == root or name.startswith(root + ".")
for name in sys.modules
for root in ("govoplan_files", "govoplan_mail")
)
print(json.dumps({
"campaign_id": first.report.campaign_id,
"built_count": first.report.built_count,
"queueable_count": first.report.queueable_count,
"normalized_eml_sha256": hashlib.sha256(first_eml).hexdigest(),
}, sort_keys=True))
"""
class CampaignExampleAcceptanceTests(unittest.TestCase):
def test_simple_announcement_validates_and_builds_without_mail_or_files(self) -> None:
self.assertTrue((FIXTURE_ROOT / "campaign.json").is_file())
self.assertTrue((FIXTURE_ROOT / "fixture.json").is_file())
with tempfile.TemporaryDirectory(prefix="govoplan-campaign-acceptance-", dir="/tmp") as temp_dir:
workspace = Path(temp_dir).resolve()
self.assertNotIn(REPOSITORY_ROOT, workspace.parents)
isolated_fixture = workspace / "simple-announcement"
shutil.copytree(FIXTURE_ROOT, isolated_fixture)
completed = subprocess.run(
[
sys.executable,
"-I",
"-c",
_ISOLATED_ACCEPTANCE_PROGRAM,
str(REPOSITORY_ROOT / "src"),
str(isolated_fixture),
],
cwd=workspace,
check=False,
capture_output=True,
text=True,
timeout=30,
)
self.assertEqual(completed.returncode, 0, completed.stderr or completed.stdout)
evidence = json.loads(completed.stdout)
self.assertEqual(evidence["campaign_id"], "simple-announcement")
self.assertEqual(evidence["built_count"], 1)
self.assertEqual(evidence["queueable_count"], 1)
self.assertRegex(evidence["normalized_eml_sha256"], r"^[0-9a-f]{64}$")
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,480 @@
from __future__ import annotations
import importlib.util
import json
import smtplib
import sys
from pathlib import Path
from typing import Any
import pytest
from govoplan_campaign.backend.campaign import load_campaign_config
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
RUNNER_PATH = REPOSITORY_ROOT / "dev" / "mail-testbed" / "run_campaign_acceptance.py"
FIXTURE_PATH = REPOSITORY_ROOT / "examples" / "greenmail-delivery" / "campaign.json"
def _load_runner():
spec = importlib.util.spec_from_file_location("govoplan_campaign_greenmail_acceptance", RUNNER_PATH)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
runner = _load_runner()
class _Response:
def __init__(self, status_code: int, payload: dict[str, Any]) -> None:
self.status_code = status_code
self._payload = payload
def json(self) -> dict[str, Any]:
return self._payload
class _AcceptanceClient:
def __init__(self) -> None:
self.campaign_json: dict[str, Any] | None = None
self.send_calls = 0
def post(self, path: str, **kwargs: Any) -> _Response:
if path == "/api/v1/campaigns":
self.campaign_json = kwargs["json"]["config"]
return _Response(
200,
{
"campaign": {"id": "campaign-internal"},
"version": {"id": "version-internal"},
},
)
if path.endswith("/validate"):
return _Response(200, {"ok": True, "error_count": 0, "warning_count": 0})
if path.endswith("/build"):
return _Response(
200,
{
"built_count": 1,
"build_failed_count": 0,
"queueable_count": 1,
},
)
if path.endswith("/send-now"):
self.send_calls += 1
if self.send_calls == 2:
return _Response(422, {"detail": "Already accepted"})
return _Response(
200,
{
"result": {
"attempted_count": 1,
"sent_count": 1,
"failed_count": 0,
"outcome_unknown_count": 0,
"skipped_count": 0,
"delivery_mode": "synchronous",
"results": [{"job_id": "not-retained", "status": "smtp_accepted"}],
}
},
)
if path.endswith("/append-sent"):
return _Response(
200,
{
"result": {
"pending_count": 1,
"processed_count": 1,
"appended_count": 1,
"failed_count": 0,
"skipped_count": 0,
"results": [{"job_id": "not-retained", "status": "appended"}],
}
},
)
raise AssertionError(f"unexpected POST {path}")
def get(self, path: str, **kwargs: Any) -> _Response:
if path.endswith("/report"):
return _Response(
200,
{
"cards": {
"jobs_total": 1,
"sent": 1,
"smtp_accepted": 1,
"failed": 0,
"outcome_unknown": 0,
"retryable": 0,
"needs_attention": 0,
"imap_appended": 1,
"imap_failed": 0,
},
"status_counts": {
"send": {"smtp_accepted": 1},
"imap": {"appended": 1},
},
},
)
raise AssertionError(f"unexpected GET {path}")
def _settings():
return runner.TestbedSettings(
smtp_host="127.0.0.1",
smtp_port=3025,
imap_host="127.0.0.1",
imap_port=3143,
username="campaign-test@govoplan.test",
password="local-test-password",
sender="campaign-test@govoplan.test",
recipient="campaign-test@govoplan.test",
sent_folder="Sent",
provider_timeout_seconds=5,
)
@pytest.mark.parametrize("host", ["localhost", "mail.test", "192.168.1.20", "8.8.8.8"])
def test_testbed_rejects_hostnames_and_non_loopback_addresses(host: str) -> None:
settings = _settings()
rejected = runner.TestbedSettings(
smtp_host=host,
smtp_port=settings.smtp_port,
imap_host=settings.imap_host,
imap_port=settings.imap_port,
username=settings.username,
password=settings.password,
sender=settings.sender,
recipient=settings.recipient,
sent_folder=settings.sent_folder,
provider_timeout_seconds=settings.provider_timeout_seconds,
)
with pytest.raises(runner.AcceptanceError, match="literal loopback|restricted to the loopback"):
rejected.assert_local_testbed()
@pytest.mark.parametrize("host", ["127.0.0.1", "127.8.9.10", "::1"])
def test_testbed_accepts_literal_loopback_and_preserves_it_in_profile(host: str) -> None:
settings = _settings()
accepted = runner.TestbedSettings(
smtp_host=host,
smtp_port=settings.smtp_port,
imap_host=host,
imap_port=settings.imap_port,
username=settings.username,
password=settings.password,
sender=settings.sender,
recipient=settings.recipient,
sent_folder=settings.sent_folder,
provider_timeout_seconds=settings.provider_timeout_seconds,
)
accepted.assert_local_testbed()
profile = runner._profile_payload(accepted, name="Literal loopback")
assert profile["smtp"]["host"] == host
assert profile["imap"]["host"] == host
def test_campaign_acceptance_orchestration_retains_only_profile_reference_and_safe_evidence() -> None:
client = _AcceptanceClient()
def snapshot_probe(_version_id: str):
assert client.campaign_json is not None
return client.campaign_json, {
"mail_profile_id": "profile-1",
"smtp_transport_revision": "opaque-smtp-revision",
"imap_transport_revision": "opaque-imap-revision",
"delivery": {"imap_append_sent": {"enabled": True, "folder": "Sent"}},
}
audit = {
"campaign.created": 1,
"campaign.validated": 1,
"campaign.messages_built": 1,
"campaign.sent_now": 1,
"campaign.send_now_rejected": 1,
"campaign.append_sent_enqueued": 1,
}
evidence, subject = runner.execute_campaign_scenario(
client,
{"Authorization": "not-retained"},
fixture_path=FIXTURE_PATH,
profile_id="profile-1",
settings=_settings(),
scenario="success",
snapshot_probe=snapshot_probe,
audit_probe=lambda _campaign_id, _version_id: audit,
append_sent=True,
repeat_send=True,
)
evidence["provider_verification"] = {
"inbox_increment": 1,
"sent_increment": 1,
"unique_subject_matches_in_inbox": 1,
"unique_subject_matches_in_sent": 1,
}
runner._assert_success_evidence(evidence)
runner._assert_evidence_safe(
{
"schema_version": runner.EVIDENCE_SCHEMA,
"coverage": {
"smtp_acceptance": True,
"partial_envelope_refusal": False,
"post_data_connection_loss_outcome_unknown": False,
"source_artifact_provenance": False,
"worker_restart_interruption": False,
},
"success": evidence,
},
settings=_settings(),
)
assert subject.startswith("[GovOPlaN acceptance ")
assert client.send_calls == 2
assert client.campaign_json is not None
assert client.campaign_json["server"] == {"mail_profile_id": "profile-1"}
assert "credentials" not in json.dumps(client.campaign_json).casefold()
serialized = json.dumps(evidence, sort_keys=True)
assert "not-retained" not in serialized
assert "local-test-password" not in serialized
def test_campaign_boundary_rejects_resolved_transport_material() -> None:
with pytest.raises(runner.AcceptanceError, match="forbidden transport material"):
runner.assert_campaign_boundary(
{"server": {"mail_profile_id": "profile-1"}},
{
"mail_profile_id": "profile-1",
"smtp_transport_revision": "smtp-revision",
"imap_transport_revision": "imap-revision",
"smtp": {"host": "should-not-be-here"},
},
profile_id="profile-1",
)
def test_success_projection_fails_closed_on_inconsistent_campaign_report() -> None:
evidence = {
"send": {
"attempted_count": 1,
"sent_count": 1,
"failed_count": 0,
"outcome_unknown_count": 0,
"skipped_count": 0,
"delivery_mode": "synchronous",
"statuses": {"smtp_accepted": 1},
},
"append_sent": {
"pending_count": 1,
"processed_count": 1,
"appended_count": 1,
"failed_count": 0,
"skipped_count": 0,
"statuses": {"appended": 1},
},
"report": {
"cards": {
"jobs_total": 1,
"sent": 0,
"smtp_accepted": 0,
"failed": 0,
"outcome_unknown": 0,
"needs_attention": 0,
"imap_appended": 0,
"imap_failed": 0,
},
"send_status_counts": {},
"imap_status_counts": {},
},
"provider_verification": {
"inbox_increment": 1,
"sent_increment": 1,
"unique_subject_matches_in_inbox": 1,
"unique_subject_matches_in_sent": 1,
},
"campaign_mail_boundary": {
"profile_reference_only": True,
"smtp_revision_frozen": True,
"imap_revision_frozen": True,
"resolved_transport_material_present": False,
},
}
with pytest.raises(runner.AcceptanceError, match="report does not agree"):
runner._assert_success_evidence(evidence)
def test_evidence_projection_rejects_unknown_status_keys() -> None:
with pytest.raises(runner.AcceptanceError, match="unsupported status"):
runner._send_evidence(
{
"delivery_mode": "synchronous",
"results": [{"status": "provider diagnostic: recipient@example.test"}],
}
)
with pytest.raises(runner.AcceptanceError, match="durable attempt status"):
runner._durable_state_evidence(
{
"job_count": 1,
"send_status_counts": {"sending": 1},
"attempt_status_counts": {"provider-secret": 1},
"unfinished_attempt_count": 1,
}
)
with pytest.raises(runner.AcceptanceError, match="synchronous delivery mode"):
runner._send_evidence(
{
"delivery_mode": "provider diagnostic: recipient@example.test",
"results": [],
}
)
with pytest.raises(runner.AcceptanceError, match="unsupported"):
runner._report_evidence(
{
"cards": {},
"status_counts": {
"send": {"smtp_accepted": 1, "provider-secret": 1},
"imap": {},
},
}
)
def _open_fault_smtp(endpoint):
client = smtplib.SMTP(endpoint.host, endpoint.port, timeout=5)
client.ehlo()
client.login("acceptance-user", "acceptance-password")
return client
def test_explicit_temporary_smtp_response_occurs_after_complete_data() -> None:
with runner.smtp_fault_endpoint("temporary_data_response") as endpoint:
client = _open_fault_smtp(endpoint)
try:
with pytest.raises(smtplib.SMTPDataError) as captured:
client.sendmail(
"sender@example.test",
["recipient@example.test"],
b"Subject: temporary\r\n\r\nmessage",
)
finally:
client.close()
assert captured.value.smtp_code == 451
assert endpoint.evidence() == {
"connection_count": 1,
"accepted_rcpt_commands": 1,
"refused_rcpt_commands": 0,
"data_transactions": 1,
}
def test_partial_recipient_refusal_retains_one_accepted_envelope() -> None:
with runner.smtp_fault_endpoint("partial_recipient_refusal") as endpoint:
client = _open_fault_smtp(endpoint)
try:
refused = client.sendmail(
"sender@example.test",
["accepted@example.test", "refused@example.test"],
b"Subject: partial\r\n\r\nmessage",
)
finally:
client.quit()
assert set(refused) == {"refused@example.test"}
assert endpoint.evidence() == {
"connection_count": 1,
"accepted_rcpt_commands": 1,
"refused_rcpt_commands": 1,
"data_transactions": 1,
}
def test_post_data_disconnect_is_a_real_ambiguous_protocol_boundary() -> None:
with runner.smtp_fault_endpoint("post_data_disconnect") as endpoint:
client = _open_fault_smtp(endpoint)
try:
with pytest.raises(smtplib.SMTPServerDisconnected):
client.sendmail(
"sender@example.test",
["recipient@example.test"],
b"Subject: ambiguous\r\n\r\nmessage",
)
finally:
client.close()
assert endpoint.wait_for_data(1)
assert endpoint.evidence()["data_transactions"] == 1
def test_partial_refusal_fixture_adds_a_second_distinct_recipient() -> None:
raw, _subject = runner.materialize_campaign_fixture(
FIXTURE_PATH,
profile_id="profile-1",
settings=_settings(),
scenario="partial_envelope_refusal",
run_token="0123456789ab",
additional_envelope_recipient=True,
)
recipients = raw["entries"]["inline"][0]["to"]
assert len(recipients) == 2
assert recipients[0]["email"] != recipients[1]["email"]
assert recipients[1]["email"].endswith("@govoplan.test")
def test_fixture_contains_no_transport_credentials() -> None:
raw = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
runner._assert_no_forbidden_campaign_keys(raw)
assert raw["server"] == {"mail_profile_id": "00000000-0000-4000-8000-000000000001"}
config = load_campaign_config(FIXTURE_PATH)
assert config.server.mail_profile_id == "00000000-0000-4000-8000-000000000001"
def test_fixture_composition_versions_are_complete_and_exact() -> None:
versions = {
"core": "0.1.13",
"access": "0.1.11",
"audit": "0.1.8",
"campaigns": "0.1.11",
"mail": "0.1.10",
"files": "0.1.9",
}
assert runner.required_composition_versions(FIXTURE_PATH, versions) == {
"core": "0.1.13",
"access": "0.1.11",
"audit": "0.1.8",
"campaigns": "0.1.11",
"mail": "0.1.10",
}
def test_fixture_composition_fails_closed_when_a_required_version_is_missing() -> None:
with pytest.raises(runner.AcceptanceError, match="versions are unavailable"):
runner.required_composition_versions(
FIXTURE_PATH,
{
"core": "0.1.13",
"access": "0.1.11",
"campaigns": "0.1.11",
"mail": "0.1.10",
},
)
def test_testbed_documentation_distinguishes_proven_and_open_failure_drills() -> None:
testbed = (REPOSITORY_ROOT / "dev" / "mail-testbed" / "README.md").read_text(encoding="utf-8")
runbook = (REPOSITORY_ROOT / "docs" / "CAMPAIGN_DELIVERY_RUNBOOK.md").read_text(encoding="utf-8")
assert "second ordinary send must be rejected before another provider effect" in testbed
assert "connection loss after complete DATA is frozen" in testbed
assert "dedicated OS process" in testbed
assert "Redis/Celery delivery" in runbook
assert "broker redelivery" in runbook
assert "celery_broker_redelivery" in testbed
assert "raw provider diagnostics" in runbook

View File

@@ -10,6 +10,7 @@ from govoplan_campaign.backend.db.models import (
JobSendStatus,
JobValidationStatus,
)
from govoplan_campaign.backend.capabilities import delivery_tasks_capability
from govoplan_campaign.backend.sending.jobs import (
SendJobResult,
_queue_validation_statuses,
@@ -48,6 +49,17 @@ def _job(entry_id: str, **overrides):
class CampaignQueueSelectionTests(unittest.TestCase):
def test_delivery_task_capability_configures_module_runtime_for_worker_processes(self):
registry = object()
settings = object()
context = SimpleNamespace(registry=registry, settings=settings)
with patch("govoplan_campaign.backend.runtime.configure_runtime") as configure:
capability = delivery_tasks_capability(context)
self.assertIsNotNone(capability)
configure.assert_called_once_with(registry=registry, settings=settings)
def test_selects_queueable_jobs_without_reclassifying_retry_states(self):
skipped_send = _job("1", send_status=JobSendStatus.FAILED_TEMPORARY.value)
skipped_queue = _job("2", queue_status=JobQueueStatus.PAUSED.value)
@@ -129,6 +141,8 @@ class CampaignQueueSelectionTests(unittest.TestCase):
)
snapshot = SimpleNamespace(
mail_profile_id="profile-1",
smtp_server_id="smtp-server-1",
smtp_credential_id="smtp-credential-1",
smtp_transport_revision="frozen",
delivery=SimpleNamespace(
rate_limit=SimpleNamespace(messages_per_minute=60),

View File

@@ -195,7 +195,10 @@ def test_post_queue_growth_is_rejected_before_batch_or_provider_preflight() -> N
patch("govoplan_campaign.backend.sending.jobs._ensure_campaign_execution_snapshot"),
patch("govoplan_campaign.backend.sending.jobs.effective_synchronous_send_policy", return_value=policy),
patch("govoplan_campaign.backend.sending.jobs._campaign_jobs_for_queue", return_value=initial_jobs),
patch("govoplan_campaign.backend.sending.jobs.queue_campaign_jobs", return_value=queued),
patch(
"govoplan_campaign.backend.sending.jobs.queue_campaign_jobs",
return_value=queued,
) as queue,
patch("govoplan_campaign.backend.sending.jobs._campaign_jobs_for_version", return_value=post_queue_jobs),
patch("govoplan_campaign.backend.sending.jobs._preflight_synchronous_send_batch") as batch_preflight,
):
@@ -207,6 +210,7 @@ def test_post_queue_growth_is_rejected_before_batch_or_provider_preflight() -> N
)
assert rejected.value.eligible_count == 3
assert queue.call_args.kwargs["commit_queue"] is False
batch_preflight.assert_not_called()
@@ -251,6 +255,7 @@ def test_asynchronous_mode_matches_actual_worker_availability(
assert result.worker_queue_available is workers_available
assert result.enqueued_count == expected_enqueued
assert persist.call_args.kwargs["delivery_mode"] == expected_mode
assert persist.call_args.kwargs["commit"] is True
assert enqueue.call_args.kwargs["enabled"] is workers_available
@@ -356,3 +361,45 @@ def test_batch_preflight_checks_every_message_before_provider_effects() -> None:
assert state_preflight.call_count == 2
assert input_preflight.call_count == 2
provider.send_campaign_email_bytes.assert_not_called()
def test_rejected_synchronous_preflight_rolls_back_staged_queue_before_audit() -> None:
session = Mock()
campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1")
version = SimpleNamespace(
id="version-1",
raw_json={},
locked_at=object(),
validation_summary={"ok": True},
build_summary={"built_count": 1},
)
rejection = SynchronousSendRejected(
"Preflight rejected the staged send.",
reason="batch_preflight_failed",
eligible_count=1,
)
with (
patch.object(router, "_get_campaign_for_principal"),
patch.object(router, "_require_permission"),
patch.object(router, "_get_campaign_for_tenant", return_value=campaign),
patch.object(router, "_get_version_for_tenant", return_value=version),
patch.object(router, "_require_mail_profile_use_if_needed"),
patch.object(router, "is_user_locked_version", return_value=False),
patch.object(router, "send_campaign_now", side_effect=rejection),
patch.object(router, "audit_from_principal") as audit,
pytest.raises(HTTPException) as rejected,
):
router.send_campaign_now_endpoint(
"campaign-1",
session=session,
principal=_Principal(
"campaigns:campaign:send", "campaigns:recipient:read"
), # type: ignore[arg-type]
)
assert rejected.value.status_code == 422
session.rollback.assert_called_once_with()
audit.assert_called_once()
assert audit.call_args.kwargs["action"] == "campaign.send_now_rejected"
assert audit.call_args.kwargs["commit"] is True

View File

@@ -1,6 +1,6 @@
{
"name": "@govoplan/campaign-webui",
"version": "0.1.10",
"version": "0.1.11",
"private": true,
"type": "module",
"main": "src/index.ts",

View File

@@ -1,11 +1,12 @@
import type {
ApiSettings,
MailConnectionTestResponse,
MailCredentialEnvelope,
MailImapFolderListResponse,
MailServerProfile,
MockMailboxMessageResponse
} from "@govoplan/core-webui";
import { apiFetch, apiGetList, apiPost } from "./client";
import { apiFetch, apiGetList, apiPath, apiPost } from "./client";
const profileActionEndpoints = {
smtp: "test-smtp",
@@ -16,11 +17,21 @@ const profileActionEndpoints = {
function runProfileAction<TResponse>(
settings: ApiSettings,
profileId: string,
action: keyof typeof profileActionEndpoints
action: keyof typeof profileActionEndpoints,
serverId?: string | null,
credentialId?: string | null,
campaignId?: string | null
): Promise<TResponse> {
return apiPost<TResponse>(
settings,
`/api/v1/mail/profiles/${encodeURIComponent(profileId)}/${profileActionEndpoints[action]}`
apiPath(
`/api/v1/mail/profiles/${encodeURIComponent(profileId)}/${profileActionEndpoints[action]}`,
{
server_id: serverId,
credential_id: credentialId,
campaign_id: campaignId
}
)
);
}
@@ -31,16 +42,58 @@ export async function listMailServerProfiles(settings: ApiSettings, includeInact
});
}
export async function testMailProfileSmtp(settings: ApiSettings, profileId: string): Promise<MailConnectionTestResponse> {
return runProfileAction<MailConnectionTestResponse>(settings, profileId, "smtp");
export function createCampaignMailCredential(
settings: ApiSettings,
profileId: string,
campaignId: string,
payload: {
name: string;
username: string;
password: string;
server_ids: string[];
}
): Promise<MailCredentialEnvelope> {
return apiFetch<MailCredentialEnvelope>(
settings,
apiPath(
`/api/v1/mail/profiles/${encodeURIComponent(profileId)}/campaign-credentials`,
{ campaign_id: campaignId }
),
{
method: "POST",
body: JSON.stringify(payload)
}
);
}
export async function testMailProfileImap(settings: ApiSettings, profileId: string): Promise<MailConnectionTestResponse> {
return runProfileAction<MailConnectionTestResponse>(settings, profileId, "imap");
export async function testMailProfileSmtp(
settings: ApiSettings,
profileId: string,
serverId?: string | null,
credentialId?: string | null,
campaignId?: string | null
): Promise<MailConnectionTestResponse> {
return runProfileAction<MailConnectionTestResponse>(settings, profileId, "smtp", serverId, credentialId, campaignId);
}
export async function listMailProfileImapFolders(settings: ApiSettings, profileId: string): Promise<MailImapFolderListResponse> {
return runProfileAction<MailImapFolderListResponse>(settings, profileId, "folders");
export async function testMailProfileImap(
settings: ApiSettings,
profileId: string,
serverId?: string | null,
credentialId?: string | null,
campaignId?: string | null
): Promise<MailConnectionTestResponse> {
return runProfileAction<MailConnectionTestResponse>(settings, profileId, "imap", serverId, credentialId, campaignId);
}
export async function listMailProfileImapFolders(
settings: ApiSettings,
profileId: string,
serverId?: string | null,
credentialId?: string | null,
campaignId?: string | null
): Promise<MailImapFolderListResponse> {
return runProfileAction<MailImapFolderListResponse>(settings, profileId, "folders", serverId, credentialId, campaignId);
}
export async function getMockMailboxMessage(settings: ApiSettings, id: string): Promise<MockMailboxMessageResponse> {

View File

@@ -8,6 +8,7 @@ import { Button } from "@govoplan/core-webui";
import { StatusBadge } from "@govoplan/core-webui";
import { PageTitle } from "@govoplan/core-webui";
import { LoadingFrame } from "@govoplan/core-webui";
import { PageScrollViewport } from "@govoplan/core-webui";
import { DismissibleAlert, TableActionGroup, i18nMessage, useGuardedNavigate } from "@govoplan/core-webui";
import { DataGrid, type DataGridColumn } from "@govoplan/core-webui";
import { createNewCampaign, listCampaignsDelta, type CampaignDeltaResponse } from "../../api/campaigns";
@@ -138,6 +139,7 @@ export default function CampaignListPage({ settings }: {settings: ApiSettings;})
return (
<PageScrollViewport>
<div className="content-pad workspace-data-page campaigns-page">
{error && <DismissibleAlert tone="danger" resetKey={error} floating>{error}</DismissibleAlert>}
@@ -179,7 +181,8 @@ export default function CampaignListPage({ settings }: {settings: ApiSettings;})
}
</LoadingFrame>
</Card>
</div>);
</div>
</PageScrollViewport>);
}

View File

@@ -2,21 +2,26 @@ import { useEffect, useState } from "react";
import {
Button,
Card,
Dialog,
DismissibleAlert,
FormField,
LoadingFrame,
MailServerFolderLookupResultView,
MetricCard,
PageTitle,
PasswordField,
ToggleSwitch,
usePlatformModuleInstalled,
usePlatformUiCapability,
type MailCredentialEnvelope,
type MailProfilesUiCapability,
type MailServerConnectionTestResult,
type MailServerEndpoint,
type MailServerFolderLookupResult
} from "@govoplan/core-webui";
import type { ApiSettings } from "../../types";
import {
createCampaignMailCredential,
listMailProfileImapFolders,
listMailServerProfiles,
testMailProfileImap,
@@ -39,6 +44,14 @@ type MailSettingsPageProps = {
view?: MailSettingsView;
};
type CampaignCredentialDraft = {
name: string;
username: string;
password: string;
useSmtp: boolean;
useImap: boolean;
};
export default function MailSettingsPage({ settings, campaignId, view = "settings" }: MailSettingsPageProps) {
const mailModuleInstalled = usePlatformModuleInstalled("mail");
const mailProfilesUi = usePlatformUiCapability<MailProfilesUiCapability>("mail.profiles");
@@ -53,6 +66,15 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
const [smtpTestResult, setSmtpTestResult] = useState<MailServerConnectionTestResult | null>(null);
const [imapTestResult, setImapTestResult] = useState<MailServerConnectionTestResult | null>(null);
const [folderResult, setFolderResult] = useState<MailServerFolderLookupResult | null>(null);
const [credentialDialogOpen, setCredentialDialogOpen] = useState(false);
const [credentialSaving, setCredentialSaving] = useState(false);
const [credentialDraft, setCredentialDraft] = useState<CampaignCredentialDraft>({
name: "",
username: "",
password: "",
useSmtp: true,
useImap: false
});
const version = data.currentVersion;
const locked = isAuditLockedVersion(version, data.campaign?.current_version_id);
@@ -78,10 +100,28 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
const server = asRecord(displayDraft.server);
const selectedProfileId = getText(server, "mail_profile_id");
const selectedProfile = mailProfiles.find((profile) => profile.id === selectedProfileId) ?? null;
const smtpServers = (selectedProfile?.servers ?? []).filter((item) => item.protocol === "smtp" && item.is_active);
const imapServers = (selectedProfile?.servers ?? []).filter((item) => item.protocol === "imap" && item.is_active);
const selectedSmtpServer = smtpServers.find((item) => item.id === getText(server, "smtp_server_id"))
?? smtpServers.find((item) => item.is_default)
?? smtpServers[0]
?? null;
const selectedImapServer = imapServers.find((item) => item.id === getText(server, "imap_server_id"))
?? imapServers.find((item) => item.is_default)
?? imapServers[0]
?? null;
const selectedSmtpCredential = selectedSmtpServer?.credentials.find((item) => item.id === getText(server, "smtp_credential_id"))
?? selectedSmtpServer?.credentials.find((item) => item.is_default)
?? selectedSmtpServer?.credentials[0]
?? null;
const selectedImapCredential = selectedImapServer?.credentials.find((item) => item.id === getText(server, "imap_credential_id"))
?? selectedImapServer?.credentials.find((item) => item.is_default)
?? selectedImapServer?.credentials[0]
?? null;
const delivery = asRecord(displayDraft.delivery);
const imapAppend = asRecord(delivery.imap_append_sent);
const imapAppendEnabled = getBool(imapAppend, "enabled");
const selectedProfileHasImap = Boolean(selectedProfile?.imap);
const selectedProfileHasImap = Boolean(selectedImapServer);
const selectedProfileUnavailable = Boolean(selectedProfileId && !profilesLoading && !selectedProfile);
const canSave = dirty && !locked && Boolean(draft) && (!migrationRequired || Boolean(selectedProfileId));
@@ -117,22 +157,140 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
function selectMailProfile(profileId: string) {
if (!mailModuleInstalled || locked) return;
patch(["server"], profileId ? { mail_profile_id: profileId } : {});
const profile = mailProfiles.find((item) => item.id === profileId);
const smtpServer = profile?.servers?.find((item) => item.protocol === "smtp" && item.is_active && item.is_default)
?? profile?.servers?.find((item) => item.protocol === "smtp" && item.is_active);
const imapServer = profile?.servers?.find((item) => item.protocol === "imap" && item.is_active && item.is_default)
?? profile?.servers?.find((item) => item.protocol === "imap" && item.is_active);
const smtpCredential = smtpServer?.credentials.find((item) => item.is_active && item.is_default)
?? smtpServer?.credentials.find((item) => item.is_active);
const imapCredential = imapServer?.credentials.find((item) => item.is_active && item.is_default)
?? imapServer?.credentials.find((item) => item.is_active);
patch(["server"], profile ? mailReference(
profile.id,
smtpServer?.id,
smtpCredential?.id,
imapServer?.id,
imapCredential?.id
) : {});
setSmtpTestResult(null);
setImapTestResult(null);
setFolderResult(null);
setProfileError("");
}
function selectServer(protocol: "smtp" | "imap", serverId: string) {
if (!selectedProfile || locked) return;
const availableServers = protocol === "smtp" ? smtpServers : imapServers;
const selected = availableServers.find((item) => item.id === serverId) ?? null;
const credential = selected?.credentials.find((item) => item.is_active && item.is_default)
?? selected?.credentials.find((item) => item.is_active)
?? null;
patch(["server"], mailReference(
selectedProfile.id,
protocol === "smtp" ? selected?.id : selectedSmtpServer?.id,
protocol === "smtp" ? credential?.id : selectedSmtpCredential?.id,
protocol === "imap" ? selected?.id : selectedImapServer?.id,
protocol === "imap" ? credential?.id : selectedImapCredential?.id
));
if (protocol === "smtp") setSmtpTestResult(null);
else {
setImapTestResult(null);
setFolderResult(null);
}
}
function selectCredential(protocol: "smtp" | "imap", credentialId: string) {
if (!selectedProfile || locked) return;
patch(["server"], mailReference(
selectedProfile.id,
selectedSmtpServer?.id,
protocol === "smtp" ? credentialId : selectedSmtpCredential?.id,
selectedImapServer?.id,
protocol === "imap" ? credentialId : selectedImapCredential?.id
));
if (protocol === "smtp") setSmtpTestResult(null);
else {
setImapTestResult(null);
setFolderResult(null);
}
}
function openCredentialDialog() {
setCredentialDraft({
name: `${data.campaign?.name || "Campaign"} credential`,
username: "",
password: "",
useSmtp: Boolean(selectedSmtpServer),
useImap: Boolean(selectedImapServer)
});
setCredentialDialogOpen(true);
setLocalError("");
}
async function saveCampaignCredential() {
if (!selectedProfile || credentialSaving) return;
const serverIds = [
credentialDraft.useSmtp ? selectedSmtpServer?.id : null,
credentialDraft.useImap ? selectedImapServer?.id : null
].filter((value): value is string => Boolean(value));
if (
!credentialDraft.name.trim() ||
!credentialDraft.username.trim() ||
!credentialDraft.password ||
serverIds.length === 0
) return;
setCredentialSaving(true);
setLocalError("");
try {
const credential = await createCampaignMailCredential(
settings,
selectedProfile.id,
campaignId,
{
name: credentialDraft.name.trim(),
username: credentialDraft.username.trim(),
password: credentialDraft.password,
server_ids: serverIds
}
);
patch(["server"], mailReference(
selectedProfile.id,
selectedSmtpServer?.id,
credentialDraft.useSmtp ? credential.id : selectedSmtpCredential?.id,
selectedImapServer?.id,
credentialDraft.useImap ? credential.id : selectedImapCredential?.id
));
setCredentialDialogOpen(false);
await refreshMailProfiles();
} catch (err) {
setLocalError(err instanceof Error ? err.message : String(err));
} finally {
setCredentialSaving(false);
}
}
async function runProfileTest(protocol: "smtp" | "imap") {
if (!selectedProfileId || locked) return;
setMailActionState(protocol);
setLocalError("");
try {
if (protocol === "smtp") {
setSmtpTestResult(await testMailProfileSmtp(settings, selectedProfileId));
setSmtpTestResult(await testMailProfileSmtp(
settings,
selectedProfileId,
selectedSmtpServer?.id,
selectedSmtpCredential?.id,
campaignId
));
} else {
setImapTestResult(await testMailProfileImap(settings, selectedProfileId));
setImapTestResult(await testMailProfileImap(
settings,
selectedProfileId,
selectedImapServer?.id,
selectedImapCredential?.id,
campaignId
));
}
} catch (err) {
const result = { ok: false, protocol, message: err instanceof Error ? err.message : String(err), details: {} };
@@ -148,7 +306,13 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
setMailActionState("folders");
setLocalError("");
try {
setFolderResult(await listMailProfileImapFolders(settings, selectedProfileId));
setFolderResult(await listMailProfileImapFolders(
settings,
selectedProfileId,
selectedImapServer?.id,
selectedImapCredential?.id,
campaignId
));
} catch (err) {
setFolderResult({ ok: false, protocol: "imap", message: err instanceof Error ? err.message : String(err), folders: [], details: {} });
} finally {
@@ -204,8 +368,11 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
{!isPolicyView && mailModuleInstalled && <Card
title="i18n:govoplan-campaign.reusable_mail_profile.f9c9aab1"
actions={<Button onClick={() => void refreshMailProfiles()} disabled={profilesLoading}>{profilesLoading ? "i18n:govoplan-campaign.loading.33ce4174" : "i18n:govoplan-campaign.reload_profiles.0fe100d1"}</Button>}>
<p className="muted small-note">i18n:govoplan-campaign.campaign_stores_only_this_stable_profile_referen.de554809</p>
actions={<div className="button-row compact-actions">
<Button onClick={() => void refreshMailProfiles()} disabled={profilesLoading}>{profilesLoading ? "i18n:govoplan-campaign.loading.33ce4174" : "i18n:govoplan-campaign.reload_profiles.0fe100d1"}</Button>
<Button variant="primary" onClick={openCredentialDialog} disabled={locked || profilesLoading || !selectedProfile || !selectedSmtpServer && !selectedImapServer}>Add campaign credential</Button>
</div>}>
<p className="muted small-note">The campaign stores stable references to the envelope, selected servers, and credentials.</p>
<div className="form-grid compact responsive-form-grid">
<FormField label="i18n:govoplan-campaign.profile.ff4fc027">
<select value={selectedProfileId} disabled={locked || profilesLoading} onChange={(event) => selectMailProfile(event.target.value)}>
@@ -213,16 +380,44 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
{mailProfiles.map((profile) => <option key={profile.id} value={profile.id}>{profile.name} ({profileScopeLabel(profile)})</option>)}
</select>
</FormField>
<FormField label="SMTP server">
<select value={selectedSmtpServer?.id ?? ""} disabled={locked || profilesLoading || !selectedProfile} onChange={(event) => selectServer("smtp", event.target.value)}>
<option value="">No SMTP server</option>
{smtpServers.map((item) => <option key={item.id} value={item.id}>{item.name} ({serverEndpointLabel(item.config)})</option>)}
</select>
</FormField>
<FormField label="SMTP credential">
<select value={selectedSmtpCredential?.id ?? ""} disabled={locked || profilesLoading || !selectedSmtpServer} onChange={(event) => selectCredential("smtp", event.target.value)}>
<option value="">No credential</option>
{selectedSmtpServer?.credentials.filter((item) => item.is_active).map((item) =>
<option key={item.id} value={item.id}>{credentialLabel(item)}</option>
)}
</select>
</FormField>
<FormField label="IMAP server">
<select value={selectedImapServer?.id ?? ""} disabled={locked || profilesLoading || !selectedProfile} onChange={(event) => selectServer("imap", event.target.value)}>
<option value="">No IMAP server</option>
{imapServers.map((item) => <option key={item.id} value={item.id}>{item.name} ({serverEndpointLabel(item.config)})</option>)}
</select>
</FormField>
<FormField label="IMAP credential">
<select value={selectedImapCredential?.id ?? ""} disabled={locked || profilesLoading || !selectedImapServer} onChange={(event) => selectCredential("imap", event.target.value)}>
<option value="">No credential</option>
{selectedImapServer?.credentials.filter((item) => item.is_active).map((item) =>
<option key={item.id} value={item.id}>{credentialLabel(item)}</option>
)}
</select>
</FormField>
</div>
{selectedProfileUnavailable && <DismissibleAlert tone="warning" dismissible={false}>i18n:govoplan-campaign.the_referenced_mail_profile_is_inactive_unavaila.abeebe26</DismissibleAlert>}
{selectedProfile && <div className="metric-grid inside">
<MetricCard label="i18n:govoplan-campaign.profile.ff4fc027" value={selectedProfile.name} />
<MetricCard label="i18n:govoplan-campaign.scope.4651a34e" value={profileScopeLabel(selectedProfile)} />
<MetricCard label="i18n:govoplan-campaign.smtp.efff9cca" value="i18n:govoplan-campaign.configured.668c5fff" tone="good" />
<MetricCard label="i18n:govoplan-campaign.imap.271f9ef2" value={selectedProfile.imap ? "i18n:govoplan-campaign.configured.668c5fff" : "i18n:govoplan-campaign.not_configured.811931bb"} tone={selectedProfile.imap ? "good" : "neutral"} />
<MetricCard label="i18n:govoplan-campaign.smtp.efff9cca" value={selectedSmtpServer?.name || "i18n:govoplan-campaign.not_configured.811931bb"} tone={selectedSmtpServer ? "good" : "neutral"} />
<MetricCard label="i18n:govoplan-campaign.imap.271f9ef2" value={selectedImapServer?.name || "i18n:govoplan-campaign.not_configured.811931bb"} tone={selectedImapServer ? "good" : "neutral"} />
</div>}
<div className="button-row compact-actions">
<Button onClick={() => void runProfileTest("smtp")} disabled={!selectedProfile || locked || Boolean(mailActionState)}>{mailActionState === "smtp" ? "i18n:govoplan-campaign.testing_smtp.8e9f8247" : "i18n:govoplan-campaign.test_profile_smtp.884a0e66"}</Button>
<Button onClick={() => void runProfileTest("smtp")} disabled={!selectedSmtpServer || locked || Boolean(mailActionState)}>{mailActionState === "smtp" ? "i18n:govoplan-campaign.testing_smtp.8e9f8247" : "i18n:govoplan-campaign.test_profile_smtp.884a0e66"}</Button>
<Button onClick={() => void runProfileTest("imap")} disabled={!selectedProfileHasImap || locked || Boolean(mailActionState)}>{mailActionState === "imap" ? "i18n:govoplan-campaign.testing_imap.13d255cf" : "i18n:govoplan-campaign.test_profile_imap.e1cec0e0"}</Button>
</div>
{smtpTestResult && <DismissibleAlert tone={smtpTestResult.ok ? "success" : "danger"} resetKey={`${smtpTestResult.protocol}:${smtpTestResult.message}`} floating>{smtpTestResult.message}</DismissibleAlert>}
@@ -230,6 +425,57 @@ export default function MailSettingsPage({ settings, campaignId, view = "setting
{profileError && <DismissibleAlert tone="warning" resetKey={profileError} floating>{profileError}</DismissibleAlert>}
</Card>}
<Dialog
open={credentialDialogOpen}
title="Add campaign credential"
onClose={() => setCredentialDialogOpen(false)}
closeDisabled={credentialSaving}
className="admin-dialog admin-dialog-wide adaptive-config-dialog"
footerClassName="button-row compact-actions"
footer={
<>
<Button onClick={() => setCredentialDialogOpen(false)} disabled={credentialSaving}>Cancel</Button>
<Button
variant="primary"
onClick={() => void saveCampaignCredential()}
disabled={
credentialSaving ||
!credentialDraft.name.trim() ||
!credentialDraft.username.trim() ||
!credentialDraft.password ||
!credentialDraft.useSmtp && !credentialDraft.useImap
}
>
{credentialSaving ? "Saving" : "Save credential"}
</Button>
</>
}
>
<div className="adaptive-config-form">
<section className="adaptive-config-section">
<header>
<h3>Campaign login</h3>
<p>This credential belongs to this campaign and can use only the selected mail servers.</p>
</header>
<div className="form-grid two">
<FormField label="Name">
<input value={credentialDraft.name} disabled={credentialSaving} onChange={(event) => setCredentialDraft({ ...credentialDraft, name: event.target.value })} autoFocus />
</FormField>
<FormField label="Username">
<input value={credentialDraft.username} disabled={credentialSaving} onChange={(event) => setCredentialDraft({ ...credentialDraft, username: event.target.value })} />
</FormField>
<FormField label="Password">
<PasswordField value={credentialDraft.password} onValueChange={(password) => setCredentialDraft({ ...credentialDraft, password })} disabled={credentialSaving} autoComplete="new-password" />
</FormField>
</div>
<div className="form-grid two">
{selectedSmtpServer && <ToggleSwitch checked={credentialDraft.useSmtp} disabled={credentialSaving} onChange={(useSmtp) => setCredentialDraft({ ...credentialDraft, useSmtp })} label={`Use for SMTP: ${selectedSmtpServer.name}`} />}
{selectedImapServer && <ToggleSwitch checked={credentialDraft.useImap} disabled={credentialSaving} onChange={(useImap) => setCredentialDraft({ ...credentialDraft, useImap })} label={`Use for IMAP: ${selectedImapServer.name}`} />}
</div>
</section>
</div>
</Dialog>
{!isPolicyView && <Card title="i18n:govoplan-campaign.imap_append.8c0d9e96" collapsible>
<div className="form-grid compact responsive-form-grid mail-server-form-grid">
<div className="mail-server-field-span mail-server-toggle-row mail-server-plain-toggle-row">
@@ -266,3 +512,28 @@ function profileScopeLabel(profile: MailServerProfile): string {
if (profile.scope_type === "group") return "i18n:govoplan-campaign.group.171a0606";
return "i18n:govoplan-campaign.campaign_scoped_mail_profile.9cbf3505";
}
function mailReference(
profileId: string,
smtpServerId?: string | null,
smtpCredentialId?: string | null,
imapServerId?: string | null,
imapCredentialId?: string | null
): Record<string, string> {
const value: Record<string, string> = { mail_profile_id: profileId };
if (smtpServerId) value.smtp_server_id = smtpServerId;
if (smtpServerId && smtpCredentialId) value.smtp_credential_id = smtpCredentialId;
if (imapServerId) value.imap_server_id = imapServerId;
if (imapServerId && imapCredentialId) value.imap_credential_id = imapCredentialId;
return value;
}
function serverEndpointLabel(config: MailServerEndpoint["config"]): string {
const host = typeof config.host === "string" && config.host ? config.host : "No host";
return config.port ? `${host}:${config.port}` : host;
}
function credentialLabel(credential: MailCredentialEnvelope): string {
const username = credential.public_data?.username;
return username ? `${credential.name} (${String(username)})` : credential.name;
}

View File

@@ -232,27 +232,23 @@ export default function RecipientDataPage({ settings, campaignId }: {settings: A
replaceInlineEntries(nextEntries);
}
function updateEntryAddressList(index: number, key: EntryAddressColumn["key"], addresses: MailboxAddress[]) {
updateEntry(index, (entry) => entryWithAddressList(entry, key, addresses));
}
function updateEntryAddressGroups(index: number, groups: Array<{key: AddressFieldKey;addresses: MailboxAddress[];}>) {
function saveEntryAddresses(
index: number,
values: HeaderAddressValues,
merges: EntryAddressMergeValues
) {
updateEntry(index, (entry) => {
let nextEntry = entry;
for (const group of groups) {
const currentAddresses = getEntryAddresses(nextEntry, group.key);
nextEntry = entryWithAddressList(nextEntry, group.key, dedupeAddresses([...currentAddresses, ...group.addresses]));
for (const column of recipientAddressOverlayColumns) {
if (!(column.key in values)) continue;
nextEntry = entryWithAddressList(nextEntry, column.key, values[column.key] ?? []);
if (!column.mergeKey || !(column.mergeKey in merges)) continue;
nextEntry = { ...nextEntry, [column.mergeKey]: Boolean(merges[column.mergeKey]) };
delete nextEntry[column.mergeKey.replace("merge_", "combine_")];
}
return nextEntry;
});
}
function updateEntryMerge(index: number, mergeKey: NonNullable<EntryAddressColumn["mergeKey"]>, checked: boolean) {
updateEntry(index, (entry) => {
const next = { ...entry, [mergeKey]: checked };
delete next[mergeKey.replace("merge_", "combine_")];
return next;
});
setRecipientAddressEditorIndex(null);
}
function updateEntryField(index: number, field: string, value: unknown) {
@@ -305,8 +301,13 @@ export default function RecipientDataPage({ settings, campaignId }: {settings: A
setAddressSourceImportOpen(false);
}
function updateHeaderAddressList(key: AddressFieldKey, addresses: MailboxAddress[]) {
patch(["recipients", key], key === "from" ? addresses.slice(0, 1) : addresses);
function saveHeaderAddresses(values: HeaderAddressValues) {
const nextRecipients = { ...recipientsSection };
for (const [key, addresses] of Object.entries(values) as Array<[AddressFieldKey, MailboxAddress[]]>) {
nextRecipients[key] = key === "from" ? addresses.slice(0, 1) : addresses;
}
patch(["recipients"], nextRecipients);
setHeaderAddressEditor(null);
}
async function copyHeaderAddresses(columns: EntryAddressColumn[]) {
@@ -508,9 +509,7 @@ export default function RecipientDataPage({ settings, campaignId }: {settings: A
locked={locked}
recipientsSection={recipientsSection}
entryDefaults={entryDefaults}
updateEntryAddressList={updateEntryAddressList}
updateEntryAddressGroups={updateEntryAddressGroups}
updateEntryMerge={updateEntryMerge}
onSave={(values, merges) => saveEntryAddresses(recipientAddressEditorIndex, values, merges)}
onClose={() => setRecipientAddressEditorIndex(null)} />
}
@@ -520,7 +519,7 @@ export default function RecipientDataPage({ settings, campaignId }: {settings: A
columns={headerAddressEditor.columns}
values={headerAddressValues(headerAddressEditor.columns, recipientsSection)}
locked={locked}
onAddressesChange={updateHeaderAddressList}
onSave={saveHeaderAddresses}
onClose={() => setHeaderAddressEditor(null)} />
}
@@ -534,13 +533,12 @@ type RecipientAddressEditorDialogProps = {
locked: boolean;
recipientsSection: Record<string, unknown>;
entryDefaults: Record<string, unknown>;
updateEntryAddressList: (index: number, key: EntryAddressColumn["key"], addresses: MailboxAddress[]) => void;
updateEntryAddressGroups: (index: number, groups: Array<{key: AddressFieldKey;addresses: MailboxAddress[];}>) => void;
updateEntryMerge: (index: number, mergeKey: NonNullable<EntryAddressColumn["mergeKey"]>, checked: boolean) => void;
onSave: (values: HeaderAddressValues, merges: EntryAddressMergeValues) => void;
onClose: () => void;
};
type HeaderAddressValues = Partial<Record<AddressFieldKey, MailboxAddress[]>>;
type EntryAddressMergeValues = Partial<Record<NonNullable<EntryAddressColumn["mergeKey"]>, boolean>>;
type AddressHeaderControlProps = {
columns: EntryAddressColumn[];
@@ -596,26 +594,33 @@ type HeaderAddressEditorDialogProps = {
columns: EntryAddressColumn[];
values: HeaderAddressValues;
locked: boolean;
onAddressesChange: (key: AddressFieldKey, addresses: MailboxAddress[]) => void;
onSave: (values: HeaderAddressValues) => void;
onClose: () => void;
};
function HeaderAddressEditorDialog({ title, columns, values, locked, onAddressesChange, onClose }: HeaderAddressEditorDialogProps) {
function HeaderAddressEditorDialog({ title, columns, values, locked, onSave, onClose }: HeaderAddressEditorDialogProps) {
const { translateText } = usePlatformLanguage();
const columnKeys = new Set(columns.map((column) => column.key));
const [draftValues, setDraftValues] = useState<HeaderAddressValues>(() => cloneAddressValues(columns, values));
const [validationError, setValidationError] = useState("");
function applyPastedAddresses(targetKey: AddressFieldKey, text: string): boolean {
const groups = parsePastedAddressGroups(targetKey, text).filter((group) => columnKeys.has(group.key));
if (groups.length === 0) return false;
const nextValues: HeaderAddressValues = { ...values };
for (const group of groups) {
nextValues[group.key] = dedupeAddresses([...(nextValues[group.key] ?? []), ...group.addresses]);
}
for (const group of groups) {
onAddressesChange(group.key, nextValues[group.key] ?? []);
}
setDraftValues((current) => mergeAddressGroups(current, groups));
setValidationError("");
return true;
}
function save() {
const prepared = prepareAddressValues(columns, draftValues, translateText);
if (prepared.error) {
setValidationError(prepared.error);
return;
}
onSave(prepared.values);
}
return (
<Dialog
open
@@ -623,17 +628,24 @@ function HeaderAddressEditorDialog({ title, columns, values, locked, onAddresses
className="recipient-address-editor-modal"
bodyClassName="recipient-address-editor-body"
onClose={onClose}
footer={<Button onClick={onClose}>i18n:govoplan-campaign.close.bbfa773e</Button>}>
footer={<>
<Button onClick={onClose}>i18n:govoplan-campaign.cancel.77dfd213</Button>
<Button variant="primary" disabled={locked} onClick={save}>i18n:govoplan-campaign.save.efc007a3</Button>
</>}>
<div className="recipient-address-editor">
{validationError && <DismissibleAlert tone="danger" dismissible={false}>{validationError}</DismissibleAlert>}
{columns.map((column) =>
<RecipientAddressCategoryEditor
key={column.key}
column={column}
addresses={values[column.key] ?? []}
addresses={draftValues[column.key] ?? []}
merge={false}
locked={locked}
onAddressesChange={(addresses) => onAddressesChange(column.key, addresses)}
onAddressesChange={(addresses) => {
setDraftValues((current) => ({ ...current, [column.key]: addresses }));
setValidationError("");
}}
onPasteAddresses={applyPastedAddresses} />
)}
@@ -647,21 +659,41 @@ function RecipientAddressEditorDialog({
locked,
recipientsSection,
entryDefaults,
updateEntryAddressList,
updateEntryAddressGroups,
updateEntryMerge,
onSave,
onClose
}: RecipientAddressEditorDialogProps) {
const { translateText } = usePlatformLanguage();
const availableColumns = recipientAddressOverlayColumns.filter((column) => entryAddressEnabled(recipientsSection, column.key));
const availableKeys = new Set(availableColumns.map((column) => column.key));
const [draftValues, setDraftValues] = useState<HeaderAddressValues>(() =>
Object.fromEntries(availableColumns.map((column) => [column.key, getEntryAddresses(entry, column.key).map(cloneMailboxAddress)]))
);
const [draftMerges, setDraftMerges] = useState<EntryAddressMergeValues>(() =>
Object.fromEntries(
availableColumns
.filter((column) => column.mergeKey)
.map((column) => [column.mergeKey!, getEntryMerge(entry, entryDefaults, column.mergeKey!)])
)
);
const [validationError, setValidationError] = useState("");
function applyPastedAddresses(targetKey: AddressFieldKey, text: string): boolean {
const groups = parsePastedAddressGroups(targetKey, text).filter((group) => availableKeys.has(group.key));
if (groups.length === 0) return false;
updateEntryAddressGroups(index, groups);
setDraftValues((current) => mergeAddressGroups(current, groups));
setValidationError("");
return true;
}
function save() {
const prepared = prepareAddressValues(availableColumns, draftValues, translateText);
if (prepared.error) {
setValidationError(prepared.error);
return;
}
onSave(prepared.values, draftMerges);
}
return (
<Dialog
open
@@ -669,9 +701,13 @@ function RecipientAddressEditorDialog({
className="recipient-address-editor-modal"
bodyClassName="recipient-address-editor-body"
onClose={onClose}
footer={<Button onClick={onClose}>i18n:govoplan-campaign.close.bbfa773e</Button>}>
footer={<>
<Button onClick={onClose}>i18n:govoplan-campaign.cancel.77dfd213</Button>
<Button variant="primary" disabled={locked} onClick={save}>i18n:govoplan-campaign.save.efc007a3</Button>
</>}>
<div className="recipient-address-editor">
{validationError && <DismissibleAlert tone="danger" dismissible={false}>{validationError}</DismissibleAlert>}
{availableColumns.length === 0 &&
<p className="muted">No individual recipient address fields are enabled.</p>
}
@@ -679,11 +715,16 @@ function RecipientAddressEditorDialog({
<RecipientAddressCategoryEditor
key={column.key}
column={column}
addresses={getEntryAddresses(entry, column.key)}
merge={column.mergeKey ? getEntryMerge(entry, entryDefaults, column.mergeKey) : false}
addresses={draftValues[column.key] ?? []}
merge={column.mergeKey ? Boolean(draftMerges[column.mergeKey]) : false}
locked={locked}
onAddressesChange={(addresses) => updateEntryAddressList(index, column.key, addresses)}
onMergeChange={column.mergeKey ? (merge) => updateEntryMerge(index, column.mergeKey!, merge) : undefined}
onAddressesChange={(addresses) => {
setDraftValues((current) => ({ ...current, [column.key]: addresses }));
setValidationError("");
}}
onMergeChange={column.mergeKey ? (merge) => {
setDraftMerges((current) => ({ ...current, [column.mergeKey!]: merge }));
} : undefined}
onPasteAddresses={applyPastedAddresses} />
)}
@@ -716,7 +757,7 @@ function RecipientAddressCategoryEditor({ column, addresses, merge, locked, onAd
function commitAddresses(nextAddresses: MailboxAddress[]) {
setDraftAddresses(nextAddresses);
onAddressesChange(nextAddresses.filter((address) => String(address.email ?? "").trim() || String(address.name ?? "").trim()));
onAddressesChange(nextAddresses);
}
function patchAddress(addressIndex: number, patch: Partial<MailboxAddress>) {
@@ -731,7 +772,7 @@ function RecipientAddressCategoryEditor({ column, addresses, merge, locked, onAd
{ name: "", email: "" },
...draftAddresses.slice(insertIndex)];
setDraftAddresses(nextAddresses);
commitAddresses(nextAddresses);
}
function removeAddress(addressIndex: number) {
@@ -2147,9 +2188,59 @@ function getAddressColumn(key: AddressFieldKey): EntryAddressColumn {
return column;
}
function cloneMailboxAddress(address: MailboxAddress): MailboxAddress {
return { name: address.name ?? "", email: address.email ?? "" };
}
function cloneAddressValues(columns: EntryAddressColumn[], values: HeaderAddressValues): HeaderAddressValues {
return Object.fromEntries(
columns.map((column) => [column.key, (values[column.key] ?? []).map(cloneMailboxAddress)])
) as HeaderAddressValues;
}
function mergeAddressGroups(
values: HeaderAddressValues,
groups: Array<{key: AddressFieldKey;addresses: MailboxAddress[];}>
): HeaderAddressValues {
const nextValues = cloneAddressValues(recipientAddressOverlayColumns, values);
for (const group of groups) {
nextValues[group.key] = dedupeAddresses([
...(nextValues[group.key] ?? []),
...group.addresses.map(cloneMailboxAddress)
]);
}
return nextValues;
}
function prepareAddressValues(
columns: EntryAddressColumn[],
values: HeaderAddressValues,
translateText: (value: string) => string
): {values: HeaderAddressValues;error: string;} {
const prepared: HeaderAddressValues = {};
for (const column of columns) {
const addresses: MailboxAddress[] = [];
for (const address of values[column.key] ?? []) {
const name = String(address.name ?? "").trim();
const email = String(address.email ?? "").trim();
if (!name && !email) continue;
if (!email) {
return {
values: prepared,
error: `${translateText(column.label)}: enter an email address or remove the incomplete row.`
};
}
addresses.push({ name, email });
}
prepared[column.key] = column.allowMultiple ? dedupeAddresses(addresses) : addresses.slice(0, 1);
}
return { values: prepared, error: "" };
}
function normalizeEditorAddressRows(column: EntryAddressColumn, addresses: MailboxAddress[]): MailboxAddress[] {
if (column.allowMultiple) return addresses;
return addresses.length > 0 ? addresses.slice(0, 1) : [{ name: "", email: "" }];
const rows = addresses.map(cloneMailboxAddress);
if (column.allowMultiple) return rows;
return rows.length > 0 ? rows.slice(0, 1) : [{ name: "", email: "" }];
}
function formatMailboxAddress(address: MailboxAddress): string {

View File

@@ -1,13 +1,26 @@
export function campaignMailProfileReferenceOnly(value: Record<string, unknown>): Record<string, unknown> {
const server = isRecord(value.server) ? value.server : {};
const rawProfileId = server.mail_profile_id;
const profileId = typeof rawProfileId === "string" ? rawProfileId.trim() : "";
const profileId = textId(server.mail_profile_id);
const smtpServerId = textId(server.smtp_server_id);
const smtpCredentialId = textId(server.smtp_credential_id);
const imapServerId = textId(server.imap_server_id);
const imapCredentialId = textId(server.imap_credential_id);
const reference: Record<string, string> = {};
if (profileId) reference.mail_profile_id = profileId;
if (profileId && smtpServerId) reference.smtp_server_id = smtpServerId;
if (profileId && smtpServerId && smtpCredentialId) reference.smtp_credential_id = smtpCredentialId;
if (profileId && imapServerId) reference.imap_server_id = imapServerId;
if (profileId && imapServerId && imapCredentialId) reference.imap_credential_id = imapCredentialId;
return {
...value,
server: profileId ? { mail_profile_id: profileId } : {}
server: reference
};
}
function textId(value: unknown): string {
return typeof value === "string" ? value.trim() : "";
}
function isRecord(value: unknown): value is Record<string, unknown> {
return Boolean(value) && typeof value === "object" && !Array.isArray(value);
}

View File

@@ -29,6 +29,7 @@ import { DismissibleAlert } from "@govoplan/core-webui";
import { LoadingFrame } from "@govoplan/core-webui";
import { MetricCard } from "@govoplan/core-webui";
import { PageTitle } from "@govoplan/core-webui";
import { PageScrollViewport } from "@govoplan/core-webui";
import {
StatusBadge,
TableActionGroup,
@@ -755,6 +756,7 @@ export default function OperatorQueuePage({ settings, auth }: {settings: ApiSett
], [navigate, permissions.canOpenReport, selectedRow, selectedVersionId]);
return (
<PageScrollViewport>
<div className="content-pad workspace-data-page">
<div className="page-heading split workspace-heading">
<div>
@@ -844,7 +846,8 @@ export default function OperatorQueuePage({ settings, auth }: {settings: ApiSett
if (cancelTarget) void runAction(cancelTarget, "cancel");
}}
onCancel={() => setCancelTarget(null)} />
</div>);
</div>
</PageScrollViewport>);
}

View File

@@ -8,6 +8,7 @@ import {
DismissibleAlert,
LoadingFrame,
MetricCard,
PageScrollViewport,
PageTitle,
StatusBadge,
TableActionGroup,
@@ -165,6 +166,7 @@ export default function AggregateReportsPage({ settings }: {settings: ApiSetting
const outcomes = report?.outcomes;
return (
<PageScrollViewport className="aggregate-reports-page">
<div className="content-pad workspace-data-page">
<div className="page-heading split workspace-heading">
<div>
@@ -246,6 +248,7 @@ export default function AggregateReportsPage({ settings }: {settings: ApiSetting
)}
</LoadingFrame>
</div>
</PageScrollViewport>
);
}

View File

@@ -2,6 +2,7 @@ import { useMemo, useState } from "react";
import { ExternalLink } from "lucide-react";
import { Button } from "@govoplan/core-webui";
import { Card } from "@govoplan/core-webui";
import { PageScrollViewport } from "@govoplan/core-webui";
import { PageTitle } from "@govoplan/core-webui";
import { FieldLabel } from "@govoplan/core-webui";
import { StatusBadge, TableActionGroup, i18nMessage } from "@govoplan/core-webui";
@@ -164,6 +165,7 @@ export default function TemplatesPage() {
}
return (
<PageScrollViewport>
<div className="content-pad workspace-data-page module-entry-page">
<div className="page-heading split workspace-heading">
<div>
@@ -194,7 +196,8 @@ export default function TemplatesPage() {
className="compact-table-wrap module-table-wrap module-entry-table" />
</Card>
</div>);
</div>
</PageScrollViewport>);
}

View File

@@ -1003,12 +1003,13 @@
.recipient-address-empty-row {
display: grid;
grid-template-columns: minmax(0, 1fr) auto;
grid-template-columns: minmax(150px, .7fr) minmax(220px, 1fr) auto;
gap: 8px;
align-items: center;
}
.recipient-address-empty {
grid-column: 1 / 3;
margin: 0;
}
@@ -1082,6 +1083,10 @@
grid-template-columns: 1fr;
}
.recipient-address-empty {
grid-column: auto;
}
.recipient-address-category-header {
align-items: flex-start;
flex-direction: column;
@@ -1218,8 +1223,8 @@
/* Shared message preview overlay --------------------------------------- */
.message-preview-modal {
width: min(920px, 100%);
height: min(780px, calc(100dvh - 48px));
width: min(1104px, calc(100vw - 48px));
height: min(936px, calc(100dvh - 48px));
max-height: calc(100dvh - 48px);
}
@@ -1237,15 +1242,16 @@
.message-preview-modal .modal-body {
flex: 1 1 auto;
min-height: 0;
display: grid;
align-content: start;
gap: 1rem;
overflow: auto;
scrollbar-gutter: stable;
overflow: hidden;
}
.message-preview-content {
display: contents;
display: flex;
flex-direction: column;
gap: 1rem;
height: 100%;
min-height: 0;
overflow: hidden;
}
.message-preview-modal .modal-footer {
@@ -1257,10 +1263,24 @@
margin-right: auto;
}
.message-preview-modal .message-display-panel {
flex: 1 1 auto;
grid-template-rows: auto minmax(0, 1fr) auto;
min-height: 0;
overflow: hidden;
}
.message-preview-modal .message-display-body-section {
grid-template-rows: auto minmax(0, 1fr);
min-height: 0;
overflow: hidden;
}
.message-preview-modal .message-display-body,
.message-preview-modal .message-display-html-frame {
height: clamp(280px, 45vh, 520px);
max-height: clamp(280px, 45vh, 520px);
height: 100%;
min-height: 0;
max-height: none;
}
.message-preview-meta {
@@ -1272,6 +1292,17 @@
margin: 0 0 0.5rem;
}
.message-preview-modal .message-display-attachments {
flex: 0 0 auto;
min-height: 0;
overflow: hidden;
}
.message-preview-modal .message-display-attachments-scroll {
max-height: 116px;
overflow: auto;
}
.attachment-chip-grid {
display: flex;
flex-wrap: wrap;
@@ -1303,6 +1334,9 @@
}
.message-preview-raw {
flex: 0 0 auto;
max-height: 132px;
overflow: auto;
border-top: 1px solid var(--line-subtle);
padding-top: 0.75rem;
}