GovOPlaN/govoplan-campaign
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Implement governed hybrid campaign delivery

Browse Source
This commit is contained in:
zemion
2026-08-02 13:58:37 +02:00
parent b38597f2be
commit 7733265cc8
40 changed files with 2531 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_postbox_delivery_orchestration.py
+79 -1
View File
@@ -11,6 +11,7 @@ from govoplan_campaign.backend.db.models import JobSendStatus
from govoplan_campaign.backend.sending.jobs import (
SendJobResult,
_MailChannelOutcome,
_PrintChannelOutcome,
_final_multichannel_status,
_send_claimed_multichannel_job,
)
@@ -49,6 +50,71 @@ class _Session:
class PostboxFallbackOrchestrationTests(unittest.TestCase):
def test_mail_unknown_never_starts_print_fallback(self) -> None:
job = SimpleNamespace(id="job-1", print_status="ready")
expected = SendJobResult(
job_id=job.id,
status=JobSendStatus.OUTCOME_UNKNOWN.value,
attempt_number=1,
)
with (
patch(
"govoplan_campaign.backend.sending.jobs._deliver_mail_channel",
return_value=_MailChannelOutcome(outcome_unknown=True),
),
patch(
"govoplan_campaign.backend.sending.jobs._deliver_print_channel"
) as deliver_print,
patch(
"govoplan_campaign.backend.sending.jobs._finalize_multichannel_job",
return_value=expected,
),
):
result = _send_claimed_multichannel_job(
_Session(job), # type: ignore[arg-type]
job=job, # type: ignore[arg-type]
claim_token="claim-1",
context=_context(), # type: ignore[arg-type]
channel_policy=DeliveryChannelPolicy.MAIL_THEN_PRINT,
use_rate_limit=False,
enqueue_imap_task=False,
)
self.assertIs(result, expected)
deliver_print.assert_not_called()
def test_mail_preacceptance_rejection_starts_print_fallback(self) -> None:
job = SimpleNamespace(id="job-1", print_status="ready")
with (
patch(
"govoplan_campaign.backend.sending.jobs._deliver_mail_channel",
return_value=_MailChannelOutcome(rejected_permanent=True),
),
patch(
"govoplan_campaign.backend.sending.jobs._deliver_print_channel",
return_value=_PrintChannelOutcome(accepted=True),
) as deliver_print,
patch(
"govoplan_campaign.backend.sending.jobs._finalize_multichannel_job",
return_value=SendJobResult(
job_id=job.id,
status=JobSendStatus.PRINT_ACCEPTED.value,
attempt_number=1,
),
),
):
_send_claimed_multichannel_job(
_Session(job), # type: ignore[arg-type]
job=job, # type: ignore[arg-type]
claim_token="claim-1",
context=_context(), # type: ignore[arg-type]
channel_policy=DeliveryChannelPolicy.MAIL_THEN_PRINT,
use_rate_limit=False,
enqueue_imap_task=False,
)
deliver_print.assert_called_once()
def test_mail_unknown_never_starts_postbox_fallback(self) -> None:
job = _job()
expected = SendJobResult(
@@ -385,6 +451,12 @@ def test_rejection_precedence_is_exhaustive_for_every_delivery_policy(
PostboxChannelOutcome(rejected_permanent=1),
JobSendStatus.PARTIALLY_ACCEPTED.value,
),
(
DeliveryChannelPolicy.PRINT,
None,
PostboxChannelOutcome(),
JobSendStatus.PRINT_ACCEPTED.value,
),
],
)
def test_accepted_delivery_decision_table(
@@ -393,7 +465,13 @@ def test_accepted_delivery_decision_table(
postbox: PostboxChannelOutcome,
expected: str,
) -> None:
assert _final_multichannel_status(channel_policy=policy, mail=mail, postbox=postbox) == expected
print_output = _PrintChannelOutcome(accepted=True) if policy == DeliveryChannelPolicy.PRINT else None
assert _final_multichannel_status(
channel_policy=policy,
mail=mail,
postbox=postbox,
print_output=print_output,
) == expected
if __name__ == "__main__":