100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.postbox import (
|
|
PostboxDeliveryCatalogRef,
|
|
PostboxDirectoryEntryRef,
|
|
PostboxDeliveryRequest,
|
|
PostboxDeliveryResult,
|
|
PostboxTargetRef,
|
|
)
|
|
from govoplan_campaign.backend.integrations import (
|
|
PostboxCampaignIntegration,
|
|
PostboxDeliveryUnavailable,
|
|
)
|
|
|
|
|
|
class _PostboxDelivery:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def deliver(self, session, request):
|
|
self.requests.append((session, request))
|
|
return PostboxDeliveryResult(
|
|
delivery_id="delivery-1",
|
|
postbox_id="postbox-1",
|
|
message_id="message-1",
|
|
address="intake@postbox",
|
|
status="accepted",
|
|
vacant=False,
|
|
holder_count=1,
|
|
)
|
|
|
|
def delivery_catalog(self, session, *, tenant_id):
|
|
return PostboxDeliveryCatalogRef()
|
|
|
|
def list_visible_postboxes(self, session, *, tenant_id, actor):
|
|
return ()
|
|
|
|
def resolve_postbox(
|
|
self,
|
|
session,
|
|
*,
|
|
tenant_id,
|
|
target,
|
|
materialize=False,
|
|
):
|
|
return PostboxDirectoryEntryRef(
|
|
id=target.postbox_id or "postbox-1",
|
|
tenant_id=tenant_id,
|
|
address="intake@postbox",
|
|
address_key="intake",
|
|
name="Intake",
|
|
status="active",
|
|
classification="internal",
|
|
)
|
|
|
|
|
|
class PostboxCampaignIntegrationTests(unittest.TestCase):
|
|
def test_optional_delivery_boundary_is_typed_and_explicit(self) -> None:
|
|
delegate = _PostboxDelivery()
|
|
integration = PostboxCampaignIntegration(delegate, delegate)
|
|
request = PostboxDeliveryRequest(
|
|
tenant_id="tenant-1",
|
|
target=PostboxTargetRef(postbox_id="postbox-1"),
|
|
producer_module="campaigns",
|
|
producer_resource_type="campaign_job",
|
|
producer_resource_id="job-1",
|
|
idempotency_key="campaign-1:job-1:postbox-1",
|
|
subject="Decision",
|
|
)
|
|
|
|
result = integration.deliver(object(), request)
|
|
|
|
self.assertTrue(integration.available)
|
|
self.assertEqual("delivery-1", result.delivery_id)
|
|
self.assertEqual(request, delegate.requests[0][1])
|
|
|
|
def test_missing_postbox_is_reported_without_importing_module_code(
|
|
self,
|
|
) -> None:
|
|
integration = PostboxCampaignIntegration()
|
|
request = PostboxDeliveryRequest(
|
|
tenant_id="tenant-1",
|
|
target=PostboxTargetRef(postbox_id="postbox-1"),
|
|
producer_module="campaigns",
|
|
producer_resource_type="campaign_job",
|
|
producer_resource_id="job-1",
|
|
idempotency_key="campaign-1:job-1:postbox-1",
|
|
subject="Decision",
|
|
)
|
|
|
|
self.assertFalse(integration.available)
|
|
with self.assertRaises(PostboxDeliveryUnavailable):
|
|
integration.deliver(object(), request)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|