user-engine/tests/test_web.py

import io
import json
import unittest
from dataclasses import replace
from datetime import timedelta
from urllib.parse import urlencode

from user_engine.adapters import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.domain import (
    OnboardingJourney, OnboardingJourneyStatus, OnboardingStep,
    OnboardingStepStatus, OnboardingTriggerType, SubsystemHandoff,
)
from user_engine.oidc import BrowserSession, OIDCClient
from user_engine.ports import IdentityDriftResult, ProvisioningResult, TenantProvisioningResult
from user_engine.service import UserEngineService
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
from user_engine.web import PortalApplication
from user_engine.domain import utc_now


SECRET = "test-proxy-secret-with-adequate-length"


def invoke(
    app, path, *, method="GET", claims=None, marker=SECRET, body=None,
    form=None, cookie=None, headers=None,
):
    payload = (
        urlencode(form).encode()
        if form is not None
        else json.dumps(body or {}).encode()
    )
    environ = {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": "",
        "CONTENT_LENGTH": str(len(payload)),
        "wsgi.input": io.BytesIO(payload),
        "HTTP_X_REQUEST_ID": "corr_test",
    }
    if form is not None:
        environ["CONTENT_TYPE"] = "application/x-www-form-urlencoded"
    if cookie is not None:
        environ["HTTP_COOKIE"] = cookie
    if claims is not None:
        environ["HTTP_X_VERIFIED_OIDC_CLAIMS"] = json.dumps(claims)
        environ["HTTP_X_USER_ENGINE_PROXY_SECRET"] = marker
    environ.update(headers or {})
    captured = {}

    def start_response(status, headers):
        captured["status"] = status
        captured["headers"] = dict(headers)

    response = b"".join(app(environ, start_response))
    return captured, response


class PortalApplicationTests(unittest.TestCase):
    def setUp(self):
        store = InMemoryUserEngineStore()
        store.migrate()
        service = UserEngineService(
            store=store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=LocalAuthorizationCheckPort(),
        )
        self.app = PortalApplication(
            service,
            trusted_proxy_secret=SECRET,
            login_url="https://kc.example/login",
        )
        self.claims = human_actor_claims(tenant="tenant:friendly:binky")

    def platform_claims(self):
        claims = human_actor_claims(subject="platform-operator", tenant="platform:root")
        claims["roles"] = ["platform-operator"]
        return claims

    def test_public_health_and_home(self):
        health, payload = invoke(self.app, "/healthz")
        self.assertEqual("200 OK", health["status"])
        self.assertEqual("no-store", health["headers"]["Cache-Control"])
        self.assertEqual("ok", json.loads(payload)["status"])
        home, html = invoke(self.app, "/")
        self.assertEqual("200 OK", home["status"])
        self.assertIn(b"Sign in with KeyCape", html)
        self.assertIn(b'name="viewport"', html)
        self.assertIn(b"focus-visible", html)
        self.assertIn(b"<main>", html)

    def test_metrics_expose_only_bounded_aggregate_state(self):
        denied, _ = invoke(self.app, "/metrics", claims={}, marker="")
        self.assertEqual("403 Forbidden", denied["status"])
        result, payload = invoke(self.app, "/metrics", claims={})
        self.assertEqual("200 OK", result["status"])
        self.assertEqual(
            "text/plain; version=0.0.4; charset=utf-8",
            result["headers"]["Content-Type"],
        )
        self.assertIn(b"user_engine_ready 1", payload)
        self.assertIn(b'user_engine_records{kind="users"} 0', payload)
        self.assertNotIn(SECRET.encode(), payload)

    def test_protected_route_rejects_untrusted_claim_header(self):
        result, payload = invoke(
            self.app, "/api/v1/me", claims=self.claims, marker="attacker"
        )
        self.assertEqual("403 Forbidden", result["status"])
        self.assertNotIn(b"attacker", payload)

    def test_expired_browser_session_and_provider_outage_fail_closed(self):
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["expired"] = BrowserSession(
            claims=self.claims, expires_at=0, csrf_token="expired-csrf"
        )
        self.app.oidc_client = oidc
        expired, _ = invoke(
            self.app, "/onboarding", cookie="ue_session=expired"
        )
        self.assertEqual("403 Forbidden", expired["status"])

        self.app.oidc_client = None
        self.app.provisioning = FailingProvisioning()
        created, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/users",
            method="POST", claims=self.claims,
            body={"primary_email": "failure@example.test", "role": "user"},
        )
        user_id = json.loads(payload)["user"]["user_id"]
        before = self.app.service.store.tenant_account(
            "tenant:friendly:binky", user_id
        ).status
        failed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            self.claims,
        )
        self.assertEqual("502 Bad Gateway", failed["status"])
        self.assertNotIn(b"provider-secret", payload)
        self.assertEqual(
            before,
            self.app.service.store.tenant_account(
                "tenant:friendly:binky", user_id
            ).status,
        )

    def test_verified_claims_create_current_user(self):
        result, payload = invoke(self.app, "/api/v1/me", claims=self.claims)
        self.assertEqual("200 OK", result["status"])
        decoded = json.loads(payload)
        self.assertEqual("tenant:friendly:binky", decoded["actor"]["tenant"])

        updated, payload = invoke_with_idempotency(
            self.app, "/api/v1/me/profile", self.claims, method="PATCH",
            body={
                "display_name": "Sample Person", "consent_accepted": True,
                "consent_version": "portal-terms-v1",
            },
        )
        self.assertEqual("200 OK", updated["status"])
        self.assertEqual("Sample Person", json.loads(payload)["display_name"])

    def test_registration_api_is_correlated(self):
        result, payload = invoke(
            self.app,
            "/api/v1/registrations",
            method="POST",
            claims=self.claims,
            body={"tenant": "tenant:friendly:binky"},
        )
        self.assertEqual("201 Created", result["status"])
        self.assertEqual("corr_test", result["headers"]["X-Request-ID"])
        self.assertEqual("factor_pending", json.loads(payload)["status"])

    def test_provision_api_links_provider_subject(self):
        self.app.provisioning = FakeProvisioning()
        created, payload = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/users",
            method="POST",
            claims=self.claims,
            body={
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("201 Created", created["status"])
        user_id = json.loads(payload)["user"]["user_id"]
        provisioned, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            claims=self.claims,
        )
        # The helper does not set an idempotency header.
        self.assertEqual("400 Bad Request", provisioned["status"])
        result, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}/provision",
            self.claims,
        )
        self.assertEqual("200 OK", result["status"])
        self.assertEqual("ada", json.loads(payload)["identity"]["subject"])
        changed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
            self.claims,
            method="PATCH",
            body={"status": "suspended"},
        )
        self.assertEqual("200 OK", changed["status"])
        self.assertEqual("suspended", json.loads(payload)["status"])
        self.assertIn(("suspend", "ada"), self.app.provisioning.actions)
        removed, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/users/{user_id}",
            self.claims, method="DELETE",
        )
        self.assertEqual("200 OK", removed["status"])
        self.assertEqual("removed", json.loads(payload)["status"])
        self.assertIn(("deprovision", "ada"), self.app.provisioning.actions)

    def test_invitation_lifecycle_is_versioned_and_replay_safe(self):
        created, payload = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "invitee@example.test", "role": "user"},
        )
        self.assertEqual("201 Created", created["status"])
        invitation = json.loads(payload)["invitation"]
        invitation_id = invitation["invitation_id"]
        self.assertEqual(1, invitation["version"])
        self.assertIsNotNone(invitation["expires_at"])

        duplicate, _ = invoke(
            self.app,
            "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "INVITEE@example.test", "role": "user"},
        )
        self.assertEqual("409 Conflict", duplicate["status"])

        listed, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            claims=self.claims,
        )
        self.assertEqual("200 OK", listed["status"])
        self.assertEqual(1, len(json.loads(payload)["items"]))

        resent, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/resend",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'},
        )
        self.assertEqual("200 OK", resent["status"])
        self.assertEqual(2, json.loads(payload)["version"])
        conflict, _ = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"1"'},
        )
        self.assertEqual("409 Conflict", conflict["status"])
        expired, payload = invoke(
            self.app,
            f"/api/v1/tenants/tenant:friendly:binky/invitations/{invitation_id}/expire",
            method="POST", claims=self.claims, headers={"HTTP_IF_MATCH": '"2"'},
        )
        self.assertEqual("200 OK", expired["status"])
        self.assertEqual("revoked", json.loads(payload)["status"])

        claimable, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "sample.user@example.test", "role": "user"},
        )
        self.assertEqual("201 Created", claimable["status"])
        claim_id = json.loads(payload)["invitation"]["invitation_id"]
        claimed, _ = invoke(
            self.app, f"/api/v1/invitations/{claim_id}/claim",
            method="POST", claims=self.claims,
        )
        self.assertEqual("200 OK", claimed["status"])
        replayed, _ = invoke(
            self.app, f"/api/v1/invitations/{claim_id}/claim",
            method="POST", claims=self.claims,
        )
        self.assertEqual("400 Bad Request", replayed["status"])

        timed, payload = invoke(
            self.app, "/api/v1/tenants/tenant:friendly:binky/invitations",
            method="POST", claims=self.claims,
            body={"primary_email": "expired@example.test", "role": "user"},
        )
        timed_id = json.loads(payload)["invitation"]["invitation_id"]
        timed_invitation = self.app.service.store.family_invitation(timed_id)
        self.app.service.store.save_family_invitation(replace(
            timed_invitation, expires_at=utc_now() - timedelta(seconds=1)
        ))
        expired_claims = dict(self.claims)
        expired_claims["email"] = "expired@example.test"
        timed_out, _ = invoke(
            self.app, f"/api/v1/invitations/{timed_id}/claim",
            method="POST", claims=expired_claims,
        )
        self.assertEqual("400 Bad Request", timed_out["status"])

    def test_outbox_delivery_replay_and_dead_letter(self):
        invoke(self.app, "/api/v1/me", claims=self.claims)
        actor = self.app.service.identity_adapter.normalize(self.claims)
        event = self.app.service.outbox_events()[0]

        def unavailable(_event):
            raise RuntimeError("notification provider unavailable; token=redacted")

        first = self.app.service.deliver_outbox(
            actor, unavailable, worker_id="worker-a", max_attempts=2
        )[0]
        self.assertEqual(1, first.delivery_attempts)
        self.assertIsNotNone(first.failed_at)
        self.assertIsNone(first.dead_lettered_at)
        replayed = self.app.service.replay_outbox(actor, event.event_id)
        self.assertIsNone(replayed.failure_reason)
        second = self.app.service.deliver_outbox(
            actor, unavailable, worker_id="worker-a", max_attempts=2
        )[0]
        self.assertIsNotNone(second.dead_lettered_at)
        self.assertLessEqual(len(second.failure_reason), 200)

        replayed = self.app.service.replay_outbox(actor, event.event_id)
        delivered = self.app.service.deliver_outbox(
            actor, lambda _event: None, worker_id="worker-b"
        )[0]
        self.assertEqual(replayed.event_id, delivered.event_id)
        self.assertIsNotNone(delivered.delivered_at)
        self.assertNotIn(delivered, self.app.service.outbox_events())

    def test_platform_tenant_bootstrap_recovery_and_outbox_transport(self):
        self.app.tenant_management = FakeTenantManagement()
        self.app.provisioning = FakeProvisioning()
        claims = self.platform_claims()
        denied, _ = invoke_with_idempotency(
            self.app, "/api/v1/platform/tenants", self.claims,
            body={"tenant": "tenant:friendly:new", "display_name": "New"},
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, payload = invoke_with_idempotency(
            self.app, "/api/v1/platform/tenants", claims,
            body={
                "tenant": "tenant:friendly:new", "display_name": "New Tenant",
                "first_admin": {"primary_email": "admin@new.test", "display_name": "Admin"},
            },
        )
        self.assertEqual("201 Created", created["status"])
        decoded = json.loads(payload)
        self.assertEqual("created", decoded["tenant"]["status"])
        self.assertEqual("tenant-admin", decoded["first_admin"]["membership"]["kind"])
        user_id = decoded["first_admin"]["user"]["user_id"]
        recovered, payload = invoke_with_idempotency(
            self.app,
            f"/api/v1/platform/tenants/tenant:friendly:new/users/{user_id}/recover",
            claims,
        )
        self.assertEqual("200 OK", recovered["status"])
        self.assertEqual("active", json.loads(payload)["tenant_account"]["status"])

        self.app.outbox_delivery = lambda _event: None
        delivered, payload = invoke(
            self.app, "/api/v1/platform/outbox/deliver", method="POST",
            claims=claims, body={"worker_id": "test-worker"},
        )
        self.assertEqual("200 OK", delivered["status"])
        self.assertTrue(json.loads(payload)["items"])

    def test_platform_browser_tenant_and_first_admin_bootstrap(self):
        claims = self.platform_claims()
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["platform"] = BrowserSession(
            claims=claims, expires_at=9999999999, csrf_token="platform-csrf",
        )
        self.app.oidc_client = oidc
        self.app.tenant_management = FakeTenantManagement()
        self.app.provisioning = FakeProvisioning()
        page, html = invoke(self.app, "/platform", cookie="ue_session=platform")
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"First administrator", html)
        denied, _ = invoke(
            self.app, "/platform/tenants", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "wrong", "tenant": "tenant:friendly:browser",
                "display_name": "Browser Tenant",
            },
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, html = invoke(
            self.app, "/platform/tenants", method="POST",
            cookie="ue_session=platform", form={
                "csrf_token": "platform-csrf", "tenant": "tenant:friendly:browser",
                "display_name": "Browser Tenant", "admin_display_name": "First Admin",
                "admin_email": "first-admin@browser.test",
            },
        )
        self.assertEqual("200 OK", created["status"])
        self.assertIn(b"awaiting onboarding", html)
        memberships = self.app.service.store.memberships_for_tenant(
            "tenant:friendly:browser"
        )
        self.assertEqual("tenant-admin", memberships[0].kind)
        admin_page, html = invoke(
            self.app, "/admin/tenant:friendly:browser", cookie="ue_session=platform"
        )
        self.assertEqual("200 OK", admin_page["status"])
        self.assertIn(b"Lifecycle diagnostics", html)
        self.assertIn(b"Recover identity", html)
        recovered, _ = invoke(
            self.app,
            f"/admin/tenant:friendly:browser/users/{memberships[0].user_id}/recover",
            method="POST", cookie="ue_session=platform",
            form={"csrf_token": "platform-csrf"},
        )
        self.assertEqual("303 See Other", recovered["status"])

    def test_admin_form_requires_csrf_and_supports_two_step_provisioning(self):
        oidc = OIDCClient(
            issuer="https://kc.example",
            client_id="portal",
            redirect_uri="https://users.example/oidc/callback",
            audience="portal",
        )
        oidc.sessions["browser"] = BrowserSession(
            claims=self.claims,
            expires_at=9999999999,
            csrf_token="csrf-test-token",
        )
        self.app.oidc_client = oidc
        self.app.provisioning = FakeProvisioning()
        denied, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "wrong",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("403 Forbidden", denied["status"])
        created, _ = invoke(
            self.app,
            "/admin/tenant:friendly:binky/users",
            method="POST",
            cookie="ue_session=browser",
            form={
                "csrf_token": "csrf-test-token",
                "display_name": "Ada Admin",
                "primary_email": "ada@example.test",
                "role": "tenant-admin",
            },
        )
        self.assertEqual("303 See Other", created["status"])
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"ada@example.test", html)
        self.assertIn(b"Create login", html)
        user_id = next(iter(self.app.service.store.users))
        handoff, html = invoke(
            self.app,
            f"/admin/tenant:friendly:binky/users/{user_id}/provision",
            method="POST",
            cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("200 OK", handoff["status"])
        self.assertIn(b"Continue to password setup", html)
        self.assertIn(b"https://kc.example/setup/password?token=opaque", html)
        page, html = invoke(
            self.app,
            "/admin/tenant:friendly:binky",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Create password setup link", html)
        self.assertIn(b"Invite a user", html)
        self.assertIn(b"Remove account", html)

    def test_browser_invitation_acceptance_and_onboarding_status(self):
        oidc = OIDCClient(
            issuer="https://kc.example", client_id="portal",
            redirect_uri="https://users.example/oidc/callback", audience="portal",
        )
        oidc.sessions["browser"] = BrowserSession(
            claims=self.claims, expires_at=9999999999, csrf_token="csrf-test-token",
        )
        self.app.oidc_client = oidc
        created, _ = invoke(
            self.app, "/admin/tenant:friendly:binky/invitations", method="POST",
            cookie="ue_session=browser", form={
                "csrf_token": "csrf-test-token", "display_name": "Invitee",
                "primary_email": "invitee@example.test", "role": "user",
            },
        )
        self.assertEqual("303 See Other", created["status"])
        invitation = next(iter(self.app.service.store.family_invitations.values()))
        page, html = invoke(
            self.app, f"/invitations/{invitation.invitation_id}",
            cookie="ue_session=browser",
        )
        self.assertEqual("200 OK", page["status"])
        self.assertIn(b"Accept invitation", html)

        wrong, _ = invoke(
            self.app, f"/invitations/{invitation.invitation_id}", method="POST",
            cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("403 Forbidden", wrong["status"])
        invitee_claims = human_actor_claims(
            subject="invitee", tenant="tenant:friendly:binky"
        )
        invitee_claims["email"] = "invitee@example.test"
        oidc.sessions["browser"] = BrowserSession(
            claims=invitee_claims, expires_at=9999999999,
            csrf_token="csrf-test-token",
        )
        accepted, _ = invoke(
            self.app, f"/invitations/{invitation.invitation_id}", method="POST",
            cookie="ue_session=browser", form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("303 See Other", accepted["status"])
        self.assertEqual("/onboarding", accepted["headers"]["Location"])
        own_journey = OnboardingJourney(
            tenant="tenant:friendly:binky", user_id=invitation.user_id,
            protocol_id="protocol-self", trigger_type=OnboardingTriggerType.INVITATION,
            status=OnboardingJourneyStatus.IN_PROGRESS, active_step_key="profile-review",
            steps=(OnboardingStep(
                step_key="profile-review", title="Review your profile",
                subsystem="user-engine", status=OnboardingStepStatus.IN_PROGRESS,
            ),),
        )
        provider_journey = OnboardingJourney(
            tenant="tenant:friendly:binky", user_id=invitation.user_id,
            protocol_id="protocol-provider", trigger_type=OnboardingTriggerType.INVITATION,
            status=OnboardingJourneyStatus.BLOCKED, active_step_key="mfa",
            steps=(OnboardingStep(
                step_key="mfa", title="Enroll MFA", subsystem="key-cape",
                status=OnboardingStepStatus.BLOCKED,
                handoff=SubsystemHandoff(
                    subsystem="key-cape", status=OnboardingStepStatus.BLOCKED,
                ),
            ),),
        )
        self.app.service.store.save_onboarding_journey(own_journey)
        self.app.service.store.save_onboarding_journey(provider_journey)
        onboarding, html = invoke(
            self.app, "/onboarding", cookie="ue_session=browser"
        )
        self.assertEqual("200 OK", onboarding["status"])
        self.assertIn(b"Onboarding progress", html)
        self.assertIn(b"tenant:friendly:binky", html)
        self.assertIn(b"Profile and consent", html)
        self.assertIn(b"Mark complete", html)
        self.assertIn(b"provider-owned surface", html)
        completed, _ = invoke(
            self.app,
            f"/onboarding/{own_journey.journey_id}/steps/profile-review/complete",
            method="POST", cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("303 See Other", completed["status"])
        provider_denied, _ = invoke(
            self.app,
            f"/onboarding/{provider_journey.journey_id}/steps/mfa/complete",
            method="POST", cookie="ue_session=browser",
            form={"csrf_token": "csrf-test-token"},
        )
        self.assertEqual("403 Forbidden", provider_denied["status"])
        saved, _ = invoke(
            self.app, "/onboarding/profile", method="POST",
            cookie="ue_session=browser", form={
                "csrf_token": "csrf-test-token", "display_name": "Updated Invitee",
                "consent_accepted": "yes",
            },
        )
        self.assertEqual("303 See Other", saved["status"])
        user = self.app.service.store.user(invitation.user_id)
        self.assertEqual("Updated Invitee", user.display_name)
        self.assertEqual("invitee@example.test", user.primary_email)
        self.assertEqual("portal-terms-v1", user.consent_version)
        self.assertIsNotNone(user.profile_completed_at)


class FakeProvisioning:
    def __init__(self):
        self.actions = []

    def provision(self, request):
        self.actions.append(("provision", request.primary_email))
        return ProvisioningResult(
            provider="netkingdom-lldap",
            external_subject=request.primary_email.split("@")[0],
            status="password_setup_required",
            password_setup_url="https://kc.example/setup/password?token=opaque",
        )

    def suspend(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("suspend", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "suspended")

    def reactivate(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("reactivate", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "active")

    def deprovision(self, *, external_subject, idempotency_key, correlation_id):
        self.actions.append(("deprovision", external_subject))
        return ProvisioningResult("netkingdom-lldap", external_subject, "removed")

    def reconcile(self, request, *, external_subject, desired_status="active"):
        self.actions.append(("reconcile", external_subject))
        return IdentityDriftResult(
            "netkingdom-lldap", external_subject, "in_sync", changed=("status",)
        )


class FakeTenantManagement:
    def create_tenant(self, *, tenant, display_name, idempotency_key, correlation_id):
        return TenantProvisioningResult(tenant=tenant, status="created")


class FailingProvisioning(FakeProvisioning):
    def provision(self, request):
        raise RuntimeError("provider-secret must never escape")


def invoke_with_idempotency(app, path, claims, *, method="POST", body=None):
    payload = json.dumps(body or {}).encode()
    environ = {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": "",
        "CONTENT_LENGTH": str(len(payload)),
        "wsgi.input": io.BytesIO(payload),
        "HTTP_X_REQUEST_ID": "corr_test",
        "HTTP_X_VERIFIED_OIDC_CLAIMS": json.dumps(claims),
        "HTTP_X_USER_ENGINE_PROXY_SECRET": SECRET,
        "HTTP_IDEMPOTENCY_KEY": "test-idempotency-123456",
    }
    captured = {}
    response = b"".join(app(
        environ,
        lambda status, headers: captured.update(
            {"status": status, "headers": dict(headers)}
        ),
    ))
    return captured, response


if __name__ == "__main__":
    unittest.main()