139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from dataclasses import replace
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import Column, String, Table, create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from govoplan_core.core.mail import MailPostboxBridgeRequest
|
|
from govoplan_core.core.postbox import PostboxTargetRef
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_mail.backend.db.models import MailServerProfile
|
|
from govoplan_mail.backend.postbox_bridge import MailPostboxBridge
|
|
|
|
|
|
RAW_MESSAGE = b"""From: Ada Example <ada@example.test>
|
|
To: Clerk <clerk@example.test>
|
|
Cc: Archive <archive@example.test>
|
|
Message-ID: <bridge-1@example.test>
|
|
Subject: Submitted evidence
|
|
MIME-Version: 1.0
|
|
Content-Type: multipart/mixed; boundary=bridge
|
|
|
|
--bridge
|
|
Content-Type: text/plain; charset=utf-8
|
|
|
|
Please process the attached evidence.
|
|
--bridge
|
|
Content-Type: application/pdf
|
|
Content-Disposition: attachment; filename=evidence.pdf
|
|
Content-Transfer-Encoding: base64
|
|
|
|
UERG
|
|
--bridge--
|
|
"""
|
|
|
|
|
|
class _DeliveryProvider:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def deliver(self, session, request):
|
|
del session
|
|
self.requests.append(request)
|
|
return SimpleNamespace(
|
|
postbox_id="postbox-1",
|
|
message_id="message-1",
|
|
delivery_id="delivery-1",
|
|
duplicate=len(self.requests) > 1,
|
|
)
|
|
|
|
|
|
class MailPostboxBridgeTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
access_users = Base.metadata.tables.get("access_users")
|
|
if access_users is None:
|
|
access_users = Table(
|
|
"access_users",
|
|
Base.metadata,
|
|
Column("id", String(36), primary_key=True),
|
|
)
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[access_users, MailServerProfile.__table__],
|
|
)
|
|
self.Session = sessionmaker(
|
|
bind=self.engine,
|
|
class_=Session,
|
|
expire_on_commit=False,
|
|
)
|
|
with self.Session() as session:
|
|
session.add(
|
|
MailServerProfile(
|
|
id="profile-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
name="Inbound",
|
|
slug="inbound",
|
|
smtp_config={},
|
|
)
|
|
)
|
|
session.commit()
|
|
self.addCleanup(self.engine.dispose)
|
|
|
|
def test_bridge_parses_bounded_content_and_delegates_idempotently(self) -> None:
|
|
delivery = _DeliveryProvider()
|
|
request = MailPostboxBridgeRequest(
|
|
tenant_id="tenant-1",
|
|
target=PostboxTargetRef(postbox_id="postbox-1"),
|
|
profile_id="profile-1",
|
|
folder="INBOX",
|
|
uid="42",
|
|
uidvalidity="20260807",
|
|
raw_message=RAW_MESSAGE,
|
|
)
|
|
with self.Session() as session, patch(
|
|
"govoplan_mail.backend.postbox_bridge.postbox_delivery_provider",
|
|
return_value=delivery,
|
|
):
|
|
first = MailPostboxBridge().bridge_message(session, request)
|
|
second = MailPostboxBridge().bridge_message(session, request)
|
|
MailPostboxBridge().bridge_message(
|
|
session,
|
|
replace(request, uidvalidity="20260808"),
|
|
)
|
|
|
|
self.assertFalse(first.duplicate)
|
|
self.assertTrue(second.duplicate)
|
|
self.assertEqual(first.source_digest, second.source_digest)
|
|
self.assertEqual(
|
|
delivery.requests[0].idempotency_key,
|
|
delivery.requests[1].idempotency_key,
|
|
)
|
|
self.assertNotEqual(
|
|
delivery.requests[0].idempotency_key,
|
|
delivery.requests[2].idempotency_key,
|
|
)
|
|
bridged = delivery.requests[0]
|
|
self.assertEqual("Submitted evidence", bridged.subject)
|
|
self.assertIn("Please process", bridged.body_text)
|
|
self.assertEqual(
|
|
["sender", "to", "cc"],
|
|
[participant.kind for participant in bridged.participants],
|
|
)
|
|
self.assertEqual(1, len(bridged.attachments))
|
|
self.assertEqual("evidence.pdf", bridged.attachments[0].name)
|
|
self.assertEqual("mail_attachment", bridged.attachments[0].reference_type)
|
|
self.assertEqual("<bridge-1@example.test>", bridged.metadata["rfc_message_id"])
|
|
self.assertEqual("20260807", bridged.metadata["mailbox_uidvalidity"])
|
|
self.assertNotIn("raw_message", bridged.metadata)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|