74 lines
3.6 KiB
Python
74 lines
3.6 KiB
Python
|
"""Scoped mail migration uses synthetic identities and a mocked HTTP boundary."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import unittest
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter, DeliveryError
|
||
|
|
from user_engine.domain import OutboxEvent
|
||
|
|
from user_engine.mail_bindings import load_mail_client_bindings, mail_client_for
|
||
|
|
|
||
|
|
|
||
|
|
class MailBindingsTests(unittest.TestCase):
|
||
|
|
def test_whole_case_sensitive_globs_and_literal_clients(self):
|
||
|
|
bindings = load_mail_client_bindings('[ ["tenant:*-family", "coulomb-social"] ]')
|
||
|
|
self.assertEqual(mail_client_for("tenant:synthetic-family", bindings), "coulomb-social")
|
||
|
|
for tenant in ("tenant:platform", "tenant:Synthetic-Family", "tenant:x-family/path"):
|
||
|
|
with self.assertRaisesRegex(ValueError, "mail_binding_denied"):
|
||
|
|
mail_client_for(tenant, bindings)
|
||
|
|
|
||
|
|
def test_invalid_configuration_never_falls_back(self):
|
||
|
|
for raw in ('', '[]', '{}', '[["*", "*"]]', '[["[a]", "client"]]',
|
||
|
|
'[["*", "one"], ["*", "two"]]', '[[true, "client"]]'):
|
||
|
|
with self.assertRaisesRegex(ValueError, "mail_client_bindings_invalid"):
|
||
|
|
load_mail_client_bindings(raw)
|
||
|
|
self.assertIsNone(load_mail_client_bindings(None))
|
||
|
|
|
||
|
|
def test_conflicting_overlaps_are_denied(self):
|
||
|
|
bindings = load_mail_client_bindings('[["tenant:*", "one"], ["tenant:x", "two"]]')
|
||
|
|
with self.assertRaisesRegex(ValueError, "mail_binding_denied"):
|
||
|
|
mail_client_for("tenant:x", bindings)
|
||
|
|
|
||
|
|
def adapter(self, bindings=None):
|
||
|
|
return HTTPOutboxDeliveryAdapter(event_url="http://events.example.test",
|
||
|
|
event_bearer_token="synthetic-event-token", mail_url="http://mail.example.test",
|
||
|
|
mail_bearer_token="synthetic-mail-token", mail_client_bindings=bindings)
|
||
|
|
|
||
|
|
def event(self, kind="family_member.invited", tenant="tenant:synthetic-family"):
|
||
|
|
return OutboxEvent(event_id="evt-synthetic", event_type=kind, aggregate_id="inv-synthetic",
|
||
|
|
payload={"primary_email": "member@example.test", "invitation_id": "inv-synthetic"},
|
||
|
|
tenant=tenant, correlation_id="trace-synthetic")
|
||
|
|
|
||
|
|
def test_binding_only_changes_mail_transport_not_event_or_outbox(self):
|
||
|
|
event = self.event()
|
||
|
|
adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]')
|
||
|
|
with patch.object(adapter, "_post") as post:
|
||
|
|
adapter(event)
|
||
|
|
mail, audit = [call.args[1] for call in post.call_args_list]
|
||
|
|
self.assertEqual(mail["client_id"], "coulomb-social")
|
||
|
|
self.assertNotIn("client_id", audit)
|
||
|
|
self.assertNotIn("primary_email", audit["data"])
|
||
|
|
self.assertNotIn("client_id", event.payload)
|
||
|
|
self.assertEqual(mail["id"], event.event_id)
|
||
|
|
|
||
|
|
def test_legacy_mode_and_non_mail_events_preserve_envelopes(self):
|
||
|
|
adapter = self.adapter()
|
||
|
|
with patch.object(adapter, "_post") as post:
|
||
|
|
adapter(self.event())
|
||
|
|
self.assertNotIn("client_id", post.call_args_list[0].args[1])
|
||
|
|
adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]')
|
||
|
|
with patch.object(adapter, "_post") as post:
|
||
|
|
adapter(self.event("membership.added", "tenant:platform"))
|
||
|
|
self.assertEqual(post.call_count, 1)
|
||
|
|
|
||
|
|
def test_unadmitted_mail_tenant_contacts_neither_lane(self):
|
||
|
|
adapter = self.adapter('[["tenant:*-family", "coulomb-social"]]')
|
||
|
|
with patch.object(adapter, "_post") as post:
|
||
|
|
with self.assertRaisesRegex(DeliveryError, "mail_binding_denied"):
|
||
|
|
adapter(self.event(tenant="tenant:platform"))
|
||
|
|
post.assert_not_called()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|