user-engine/tests/test_platform_adapters.py

import io
import json
import tempfile
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)
        self.assertEqual(request["subject"]["attributes"]["tenant_roles"], ["CUS"])

    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")
        self.assertIsNone(decision.decision_id)
        self.assertEqual(decision.stance_applied, "fail_closed")

    def test_flex_auth_reads_rotating_caller_token_for_each_decision(self):
        with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as token_file:
            token_file.write("projected-token-1\n")
            token_file.flush()
            body = _Response(json.dumps({"id": "decision:1", "effect": "allow"}).encode())
            adapter = FlexAuthHTTPAdapter(
                base_url="http://flex-auth", bearer_token_file=token_file.name
            )
            with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
                adapter.check(_request())
            self.assertEqual(
                call.call_args.args[0].get_header("Authorization"),
                "Bearer projected-token-1",
            )

            token_file.seek(0)
            token_file.truncate()
            token_file.write("projected-token-2\n")
            token_file.flush()
            body = _Response(json.dumps({"id": "decision:2", "effect": "allow"}).encode())
            with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
                adapter.check(_request())
            self.assertEqual(
                call.call_args.args[0].get_header("Authorization"),
                "Bearer projected-token-2",
            )

    def test_flex_auth_fails_closed_on_unusable_caller_token(self):
        """A caller that cannot prove its identity must never reach the service."""
        with tempfile.TemporaryDirectory() as directory:
            empty = f"{directory}/empty-token"
            with open(empty, "w", encoding="utf-8") as handle:
                handle.write("   \n")
            unusable = {
                "missing": f"{directory}/absent-token",
                "empty": empty,
                "unreadable": directory,
            }
            for label, path in unusable.items():
                with self.subTest(token=label):
                    adapter = FlexAuthHTTPAdapter(
                        base_url="http://flex-auth", bearer_token_file=path
                    )
                    with patch("user_engine.adapters.flex_auth.urlopen") as call:
                        decision = adapter.check(_request())
                    call.assert_not_called()
                    self.assertEqual(decision.effect, AuthorizationEffect.DENY)
                    self.assertEqual(
                        decision.reason, "authorization service unavailable"
                    )

    def test_flex_auth_deny_reason_never_carries_the_caller_token(self):
        with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as token_file:
            token_file.write("super-secret-projected-token\n")
            token_file.flush()
            adapter = FlexAuthHTTPAdapter(
                base_url="http://flex-auth", bearer_token_file=token_file.name
            )
            with patch(
                "user_engine.adapters.flex_auth.urlopen", side_effect=URLError("down")
            ):
                decision = adapter.check(_request())
        self.assertEqual(decision.effect, AuthorizationEffect.DENY)
        self.assertNotIn("super-secret-projected-token", str(decision.reason))
        self.assertNotIn("super-secret-projected-token", repr(decision))

    def test_invitation_delivery_calls_mail_and_event_with_idempotency(self):
        adapter = HTTPOutboxDeliveryAdapter(
            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"))
        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")
        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")
        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",
            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(
        issuer="https://issuer", subject="subject-1", tenant="tenant-a",
        principal_type=PrincipalType.HUMAN, audience=("user-engine",),
        roles=("tenant-admin",), tenant_roles=("CUS",),
    )
    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()