93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import patch
|
|
from urllib.error import URLError
|
|
|
|
from user_engine.adapters.delivery import HTTPOutboxDeliveryAdapter
|
|
from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter
|
|
from user_engine.domain import (
|
|
Actor,
|
|
AuthorizationEffect,
|
|
AuthorizationRequest,
|
|
OutboxEvent,
|
|
PrincipalType,
|
|
)
|
|
|
|
|
|
class _Response(io.BytesIO):
|
|
status = 200
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
self.close()
|
|
|
|
|
|
class PlatformAdapterTests(unittest.TestCase):
|
|
def test_flex_auth_maps_allow_and_decision_id(self):
|
|
body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode())
|
|
with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
|
|
decision = FlexAuthHTTPAdapter(base_url="http://flex-auth").check(_request())
|
|
self.assertEqual(decision.effect, AuthorizationEffect.ALLOW)
|
|
self.assertEqual(decision.decision_id, "decision:1")
|
|
request = json.loads(call.call_args.args[0].data)
|
|
self.assertEqual(request["resource"]["system"], "user-engine")
|
|
self.assertEqual(request["context"]["self"], True)
|
|
|
|
def test_flex_auth_fails_closed_when_unavailable(self):
|
|
with patch("user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down")):
|
|
decision = FlexAuthHTTPAdapter(base_url="http://flex-auth").check(_request())
|
|
self.assertEqual(decision.effect, AuthorizationEffect.DENY)
|
|
self.assertEqual(decision.reason, "authorization service unavailable")
|
|
|
|
def test_invitation_delivery_calls_mail_and_event_with_idempotency(self):
|
|
adapter = HTTPOutboxDeliveryAdapter(
|
|
event_url="http://events", mail_url="http://mail", bearer_token="opaque"
|
|
)
|
|
with patch("user_engine.adapters.delivery.urlopen", return_value=_Response()) as call:
|
|
adapter(_event("family_member.invited"))
|
|
self.assertEqual([item.args[0].full_url for item in call.call_args_list],
|
|
["http://mail", "http://events"])
|
|
for item in call.call_args_list:
|
|
self.assertEqual(item.args[0].get_header("Idempotency-key"), "evt-1")
|
|
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")
|
|
self.assertNotIn("primary_email", event_payload["data"])
|
|
self.assertTrue(event_payload["data"]["recipient_present"])
|
|
|
|
def test_non_mail_event_only_calls_event_lane(self):
|
|
adapter = HTTPOutboxDeliveryAdapter(
|
|
event_url="http://events", mail_url="http://mail", bearer_token="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")
|
|
|
|
def _request():
|
|
actor = Actor(
|
|
issuer="https://issuer", subject="subject-1", tenant="tenant-a",
|
|
principal_type=PrincipalType.HUMAN, audience=("user-engine",),
|
|
roles=("tenant-admin",),
|
|
)
|
|
return AuthorizationRequest(
|
|
actor=actor, resource_type="user-engine:user", resource_id="user-1",
|
|
action="user.update", tenant="tenant-a", correlation_id="corr-1",
|
|
target_user_id="user-1", context={"self": True},
|
|
)
|
|
|
|
|
|
def _event(event_type):
|
|
return OutboxEvent(
|
|
event_id="evt-1", event_type=event_type, aggregate_id="inv-1",
|
|
payload={"primary_email": "person@example.test"}, tenant="tenant-a",
|
|
correlation_id="corr-1", occurred_at=datetime(2026, 8, 8, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|