359 lines
14 KiB
Python
359 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from govoplan_campaign.backend import router
|
|
from govoplan_campaign.backend.delivery_policy import (
|
|
CampaignDeliveryPolicyError,
|
|
DEFAULT_SYNCHRONOUS_SEND_MAX_RECIPIENT_JOBS,
|
|
effective_synchronous_send_policy,
|
|
)
|
|
from govoplan_campaign.backend.db.models import (
|
|
JobBuildStatus,
|
|
JobQueueStatus,
|
|
JobSendStatus,
|
|
JobValidationStatus,
|
|
)
|
|
from govoplan_campaign.backend.sending.jobs import (
|
|
QueueCampaignResult,
|
|
SynchronousSendRejected,
|
|
_ensure_synchronous_send_count_allowed,
|
|
_preflight_synchronous_send_batch,
|
|
queue_campaign_jobs,
|
|
send_campaign_now,
|
|
synchronous_send_candidate_jobs,
|
|
synchronous_send_options,
|
|
)
|
|
|
|
|
|
class _PolicySession:
|
|
def __init__(self, settings: dict[str, object] | None = None) -> None:
|
|
self.tenant = SimpleNamespace(settings=settings or {})
|
|
|
|
def get(self, _model, _id):
|
|
return self.tenant
|
|
|
|
|
|
def _version() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
id="version-1",
|
|
locked_at=object(),
|
|
published_at=None,
|
|
validation_summary={"ok": True},
|
|
build_summary={"build_token": "build-1"},
|
|
editor_state={
|
|
"review_send": {
|
|
"build_token": "build-1",
|
|
"inspection_complete": True,
|
|
"reviewed_message_keys": ["reviewed"],
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
class _Principal:
|
|
def __init__(self, *scopes: str) -> None:
|
|
self.scopes = set(scopes)
|
|
self.tenant_id = "tenant-1"
|
|
self.user = SimpleNamespace(id="user-1")
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope in self.scopes
|
|
|
|
|
|
def _job(job_id: str, **overrides: object) -> SimpleNamespace:
|
|
values: dict[str, object] = {
|
|
"id": job_id,
|
|
"entry_id": job_id,
|
|
"entry_index": 1,
|
|
"build_status": JobBuildStatus.BUILT.value,
|
|
"validation_status": JobValidationStatus.READY.value,
|
|
"queue_status": JobQueueStatus.DRAFT.value,
|
|
"send_status": JobSendStatus.NOT_QUEUED.value,
|
|
"eml_local_path": f"{job_id}.eml",
|
|
"eml_storage_key": None,
|
|
}
|
|
values.update(overrides)
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
def test_synchronous_policy_defaults_to_25_and_tenant_can_only_narrow() -> None:
|
|
default = effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={},
|
|
)
|
|
assert default.max_recipient_jobs == DEFAULT_SYNCHRONOUS_SEND_MAX_RECIPIENT_JOBS == 25
|
|
assert default.source == "deployment_default"
|
|
|
|
narrowed = effective_synchronous_send_policy(
|
|
_PolicySession(
|
|
{"campaign_delivery_policy": {"synchronous_send_max_recipients": 10}}
|
|
), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "40"},
|
|
)
|
|
assert narrowed.max_recipient_jobs == 10
|
|
assert narrowed.source == "tenant"
|
|
|
|
ceiling = effective_synchronous_send_policy(
|
|
_PolicySession(
|
|
{"campaign_delivery_policy": {"synchronous_send_max_recipients": 100}}
|
|
), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "40"},
|
|
)
|
|
assert ceiling.max_recipient_jobs == 40
|
|
assert ceiling.source == "deployment_ceiling"
|
|
|
|
|
|
@pytest.mark.parametrize("value", [True, -1, 501, "2.5", "unbounded"])
|
|
def test_invalid_synchronous_policy_fails_closed(value: object) -> None:
|
|
with pytest.raises(CampaignDeliveryPolicyError):
|
|
effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": value}, # type: ignore[dict-item]
|
|
)
|
|
|
|
|
|
def test_candidate_count_uses_exact_built_and_reviewed_job_states() -> None:
|
|
jobs = [
|
|
_job("ready"),
|
|
_job("queued", queue_status=JobQueueStatus.QUEUED.value, send_status=JobSendStatus.QUEUED.value),
|
|
_job("reviewed", validation_status=JobValidationStatus.NEEDS_REVIEW.value),
|
|
_job("blocked", validation_status=JobValidationStatus.BLOCKED.value),
|
|
_job("failed", send_status=JobSendStatus.FAILED_TEMPORARY.value),
|
|
_job("missing-eml", eml_local_path=None),
|
|
]
|
|
|
|
candidates = synchronous_send_candidate_jobs(_version(), jobs) # type: ignore[arg-type]
|
|
|
|
assert [job.id for job in candidates] == ["ready", "queued", "reviewed"]
|
|
|
|
|
|
def test_limit_and_zero_count_reject_before_delivery() -> None:
|
|
policy = effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "2"},
|
|
)
|
|
with pytest.raises(SynchronousSendRejected, match="No eligible") as empty:
|
|
_ensure_synchronous_send_count_allowed(0, policy=policy)
|
|
assert empty.value.reason == "no_eligible_recipient_jobs"
|
|
|
|
with pytest.raises(SynchronousSendRejected, match="Queue it") as oversized:
|
|
_ensure_synchronous_send_count_allowed(3, policy=policy)
|
|
assert oversized.value.audit_details()["eligible_recipient_job_count"] == 3
|
|
assert oversized.value.audit_details()["synchronous_send_policy"]["max_recipient_jobs"] == 2
|
|
|
|
|
|
def test_exact_synchronous_policy_boundary_is_allowed() -> None:
|
|
policy = effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "2"},
|
|
)
|
|
|
|
_ensure_synchronous_send_count_allowed(2, policy=policy)
|
|
|
|
|
|
def test_post_queue_growth_is_rejected_before_batch_or_provider_preflight() -> None:
|
|
policy = effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={"GOVOPLAN_CAMPAIGN_SYNCHRONOUS_SEND_MAX_RECIPIENTS": "2"},
|
|
)
|
|
campaign = SimpleNamespace(id="campaign-1", current_version_id="version-1")
|
|
initial_jobs = [_job("one"), _job("two")]
|
|
post_queue_jobs = [
|
|
_job(
|
|
job_id,
|
|
queue_status=JobQueueStatus.QUEUED.value,
|
|
send_status=JobSendStatus.QUEUED.value,
|
|
)
|
|
for job_id in ("one", "two", "concurrent")
|
|
]
|
|
queued = QueueCampaignResult(
|
|
campaign_id="campaign-1",
|
|
version_id="version-1",
|
|
queued_count=2,
|
|
skipped_count=0,
|
|
blocked_count=0,
|
|
enqueued_count=0,
|
|
delivery_mode="synchronous",
|
|
)
|
|
|
|
with (
|
|
patch("govoplan_campaign.backend.sending.jobs._get_campaign_for_tenant", return_value=campaign),
|
|
patch("govoplan_campaign.backend.sending.jobs._get_current_version", return_value=_version()),
|
|
patch("govoplan_campaign.backend.sending.jobs._ensure_version_validated_and_locked"),
|
|
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._campaign_jobs_for_version", return_value=post_queue_jobs),
|
|
patch("govoplan_campaign.backend.sending.jobs._preflight_synchronous_send_batch") as batch_preflight,
|
|
):
|
|
with pytest.raises(SynchronousSendRejected, match="above the effective") as rejected:
|
|
send_campaign_now(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
)
|
|
|
|
assert rejected.value.eligible_count == 3
|
|
batch_preflight.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("workers_available", "expected_mode", "expected_enqueued"),
|
|
((False, "database_queue", 0), (True, "worker_queue", 1)),
|
|
)
|
|
def test_asynchronous_mode_matches_actual_worker_availability(
|
|
workers_available: bool,
|
|
expected_mode: str,
|
|
expected_enqueued: int,
|
|
) -> None:
|
|
campaign = SimpleNamespace(id="campaign-1")
|
|
version = _version()
|
|
job = _job("one")
|
|
|
|
with (
|
|
patch("govoplan_campaign.backend.sending.jobs._celery_enabled", return_value=workers_available),
|
|
patch("govoplan_campaign.backend.sending.jobs._get_campaign_for_tenant", return_value=campaign),
|
|
patch("govoplan_campaign.backend.sending.jobs._get_current_version", return_value=version),
|
|
patch("govoplan_campaign.backend.sending.jobs._ensure_version_validated_and_locked"),
|
|
patch("govoplan_campaign.backend.sending.jobs._ensure_campaign_execution_snapshot"),
|
|
patch("govoplan_campaign.backend.sending.jobs._campaign_jobs_for_queue", return_value=[job]),
|
|
patch(
|
|
"govoplan_campaign.backend.sending.jobs._select_campaign_jobs_for_queue",
|
|
return_value=([job], 0, 0),
|
|
),
|
|
patch("govoplan_campaign.backend.sending.jobs._persist_campaign_queue") as persist,
|
|
patch(
|
|
"govoplan_campaign.backend.sending.jobs._enqueue_campaign_jobs",
|
|
return_value=expected_enqueued,
|
|
) as enqueue,
|
|
):
|
|
result = queue_campaign_jobs(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
enqueue_celery=True,
|
|
)
|
|
|
|
assert result.delivery_mode == expected_mode
|
|
assert result.worker_queue_available is workers_available
|
|
assert result.enqueued_count == expected_enqueued
|
|
assert persist.call_args.kwargs["delivery_mode"] == expected_mode
|
|
assert enqueue.call_args.kwargs["enabled"] is workers_available
|
|
|
|
|
|
def test_invalid_policy_disables_synchronous_mode_without_hiding_queue_availability() -> None:
|
|
campaign = SimpleNamespace(id="campaign-1")
|
|
version = _version()
|
|
with (
|
|
patch("govoplan_campaign.backend.sending.jobs._get_campaign_for_tenant", return_value=campaign),
|
|
patch("govoplan_campaign.backend.sending.jobs._get_version_for_campaign", return_value=version),
|
|
patch("govoplan_campaign.backend.sending.jobs._campaign_jobs_for_version", return_value=[_job("one")]),
|
|
patch("govoplan_campaign.backend.sending.jobs._celery_enabled", return_value=True),
|
|
patch(
|
|
"govoplan_campaign.backend.sending.jobs.effective_synchronous_send_policy",
|
|
side_effect=CampaignDeliveryPolicyError("invalid deployment value"),
|
|
),
|
|
):
|
|
options = synchronous_send_options(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
campaign_id="campaign-1",
|
|
)
|
|
|
|
assert options["worker_queue_available"] is True
|
|
assert options["synchronous_send"]["allowed"] is False
|
|
assert options["synchronous_send"]["reason"] == "policy_configuration_invalid"
|
|
assert options["synchronous_send"]["policy"] == {}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "required_scope"),
|
|
(
|
|
("/campaigns/{campaign_id}/send-now", "campaigns:campaign:send"),
|
|
("/campaigns/{campaign_id}/queue", "campaigns:campaign:queue"),
|
|
),
|
|
)
|
|
def test_delivery_endpoints_require_their_mode_permission_and_recipient_authority(
|
|
path: str,
|
|
required_scope: str,
|
|
) -> None:
|
|
route = next(item for item in router.router.routes if item.path == path)
|
|
dependency = next(item for item in route.dependant.dependencies if item.name == "principal")
|
|
|
|
with pytest.raises(HTTPException) as missing_mode_permission:
|
|
dependency.call(_Principal("campaigns:recipient:read"))
|
|
assert missing_mode_permission.value.status_code == 403
|
|
allowed = _Principal(required_scope, "campaigns:recipient:read")
|
|
assert dependency.call(allowed) is allowed
|
|
|
|
mode_only = _Principal(required_scope)
|
|
with (
|
|
patch.object(router, "_get_campaign_for_principal"),
|
|
pytest.raises(HTTPException) as missing_recipient_authority,
|
|
):
|
|
if required_scope == "campaigns:campaign:send":
|
|
router.send_campaign_now_endpoint(
|
|
"campaign-1",
|
|
session=Mock(),
|
|
principal=mode_only, # type: ignore[arg-type]
|
|
)
|
|
else:
|
|
router.queue_campaign(
|
|
"campaign-1",
|
|
session=Mock(),
|
|
principal=mode_only, # type: ignore[arg-type]
|
|
)
|
|
assert missing_recipient_authority.value.status_code == 403
|
|
assert "campaigns:recipient:read" in missing_recipient_authority.value.detail
|
|
|
|
|
|
def test_batch_preflight_checks_every_message_before_provider_effects() -> None:
|
|
jobs = [_job("one"), _job("two")]
|
|
contexts = {
|
|
"one": SimpleNamespace(snapshot=SimpleNamespace(smtp_transport_revision="revision-1")),
|
|
"two": SimpleNamespace(snapshot=SimpleNamespace(smtp_transport_revision="revision-1")),
|
|
}
|
|
policy = effective_synchronous_send_policy(
|
|
_PolicySession(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
environ={},
|
|
)
|
|
provider = Mock()
|
|
|
|
with (
|
|
patch("govoplan_campaign.backend.sending.jobs._preflight_send_campaign_job", return_value=None) as state_preflight,
|
|
patch(
|
|
"govoplan_campaign.backend.sending.jobs._send_job_delivery_context",
|
|
side_effect=lambda _session, job: contexts[job.id],
|
|
) as input_preflight,
|
|
patch(
|
|
"govoplan_campaign.backend.sending.jobs.profile_delivery_summary",
|
|
return_value={"smtp_transport_revision": "revision-1"},
|
|
),
|
|
patch("govoplan_campaign.backend.sending.jobs.mail_integration", return_value=provider),
|
|
):
|
|
result = _preflight_synchronous_send_batch(
|
|
object(), # type: ignore[arg-type]
|
|
version=_version(), # type: ignore[arg-type]
|
|
jobs=jobs, # type: ignore[arg-type]
|
|
policy=policy,
|
|
)
|
|
|
|
assert list(result) == ["one", "two"]
|
|
assert state_preflight.call_count == 2
|
|
assert input_preflight.call_count == 2
|
|
provider.send_campaign_email_bytes.assert_not_called()
|