Add opt-in tenant-to-client bindings for invitation mail
User-engine side of EMAIL-WP-0008: USER_ENGINE_MAIL_CLIENT_BINDINGS maps tenant globs to a single client_id carried on invitation mail envelopes only. Unmatched or conflicting tenants fail as mail_binding_denied before either lane is contacted, and the outbox now records that reason instead of collapsing it to delivery_unavailable. Unset keeps legacy envelopes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 445231@bnt-lap001 Assistant-Session: 024a807e-0cc5-4f2c-bc8a-b83322d1395d
This commit is contained in:
parent
a9360842f8
commit
adf6e992a6
9 changed files with 199 additions and 2 deletions
73
tests/test_mail_bindings.py
Normal file
73
tests/test_mail_bindings.py
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
"""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()
|
||||
|
|
@ -106,6 +106,23 @@ class ServiceOperations(JourneyFixture):
|
|||
self.app.service.store.outbox_event(event.event_id).delivered_at
|
||||
)
|
||||
|
||||
def test_mail_binding_denial_is_recorded_not_collapsed(self):
|
||||
event = self.seed()
|
||||
|
||||
def deny(event):
|
||||
raise DeliveryError("mail_binding_denied")
|
||||
|
||||
self.app.outbox_delivery = deny
|
||||
self.post(
|
||||
"/platform/operations/deliver",
|
||||
who="operator",
|
||||
event_id=event.event_id,
|
||||
confirmed=True,
|
||||
)
|
||||
failed = self.app.service.store.outbox_event(event.event_id)
|
||||
self.assertEqual("mail_binding_denied", failed.failure_reason)
|
||||
self.assertIsNone(failed.delivered_at)
|
||||
|
||||
def test_crashed_claim_reconciles_under_event_guard(self):
|
||||
event = self.seed()
|
||||
self.app.service.store.save_outbox(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue