coulomb/user-engine
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Separate event and mail delivery credentials
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source
This commit is contained in:
tegwick 2026-08-13 14:56:24 +02:00
parent 9480463b24
commit a0f39f58bd
4 changed files with 45 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/test_platform_adapters.py
View file

@ -45,7 +45,8 @@ class PlatformAdapterTests(unittest.TestCase):
def test_invitation_delivery_calls_mail_and_event_with_idempotency(self):
adapter = HTTPOutboxDeliveryAdapter(
event_url="http://events", mail_url="http://mail", bearer_token="opaque"
event_url="http://events", mail_url="http://mail",
event_bearer_token="event-opaque", mail_bearer_token="mail-opaque",
)
with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call:
adapter(_event("family_member.invited"))
@ -53,6 +54,10 @@ class PlatformAdapterTests(unittest.TestCase):
["http://mail", "http://events"])
for item in call.call_args_list:
self.assertEqual(item.args[0].get_header("Idempotency-key"), "evt-1")
self.assertEqual(call.call_args_list[0].args[0].get_header("Authorization"),
"Bearer mail-opaque")
self.assertEqual(call.call_args_list[1].args[0].get_header("Authorization"),
"Bearer event-opaque")
mail_payload = json.loads(call.call_args_list[0].args[0].data)
event_payload = json.loads(call.call_args_list[1].args[0].data)
self.assertEqual(mail_payload["data"]["primary_email"], "person@example.test")
@ -61,12 +66,23 @@ class PlatformAdapterTests(unittest.TestCase):
def test_non_mail_event_only_calls_event_lane(self):
adapter = HTTPOutboxDeliveryAdapter(
event_url="http://events", mail_url="http://mail", bearer_token="opaque"
event_url="http://events", mail_url="http://mail",
event_bearer_token="event-opaque", mail_bearer_token="mail-opaque",
)
with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call:
adapter(_event("membership.added"))
self.assertEqual(call.call_count, 1)
self.assertEqual(call.call_args.args[0].full_url, "http://events")
self.assertEqual(call.call_args.args[0].get_header("Authorization"),
"Bearer event-opaque")
def test_mail_event_fails_closed_without_mail_token(self):
adapter = HTTPOutboxDeliveryAdapter(
event_url="http://events", mail_url="http://mail",
event_bearer_token="event-opaque",
)
with self.assertRaisesRegex(RuntimeError, "mail delivery token is required"):
adapter(_event("family_member.invited"))
def _request():
actor = Actor(