tenant-engine/tests/test_authz_flex.py

import httpx
import pytest

from tenant_engine.app import create_app
from tenant_engine.authz import (
    DefaultDenyWriteAuthorizer,
    FlexAuthWriteAuthorizer,
    WriteAuthorizationDeniedError,
)
from tenant_engine.config import Settings
from tenant_engine.flex_auth import FlexAuthCheckClient


def _settings(*, flex_auth_url: str | None) -> Settings:
    return Settings(
        flex_auth_base_url=flex_auth_url,
        flex_auth_timeout_seconds=1,
        host="127.0.0.1",
        port=8090,
    )


def test_create_app_defaults_to_default_deny_without_flex_auth_url() -> None:
    app = create_app(settings=_settings(flex_auth_url=None))
    assert isinstance(app.state.authorizer, DefaultDenyWriteAuthorizer)


def test_create_app_uses_flex_auth_authorizer_when_url_configured() -> None:
    app = create_app(settings=_settings(flex_auth_url="https://flex-auth.example.test"))
    assert isinstance(app.state.authorizer, FlexAuthWriteAuthorizer)


def test_flex_auth_authorizer_denies_on_deny_effect() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"id": "d-1", "effect": "deny", "resource": {}, "subject": {}, "provenance": {}})

    client = FlexAuthCheckClient(
        base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
    )
    authorizer = FlexAuthWriteAuthorizer(client=client)

    with pytest.raises(WriteAuthorizationDeniedError):
        authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops")


def test_flex_auth_authorizer_denies_on_not_applicable_effect() -> None:
    """The realistic state until FLEX-WP-0008's policy package exists."""

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200, json={"id": "d-1", "effect": "not_applicable", "resource": {}, "subject": {}, "provenance": {}}
        )

    client = FlexAuthCheckClient(
        base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
    )
    authorizer = FlexAuthWriteAuthorizer(client=client)

    with pytest.raises(WriteAuthorizationDeniedError):
        authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops")


def test_flex_auth_authorizer_allows_on_allow_effect() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})

    client = FlexAuthCheckClient(
        base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
    )
    authorizer = FlexAuthWriteAuthorizer(client=client)

    authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops")  # does not raise


def test_full_write_lifecycle_succeeds_when_flex_auth_allows() -> None:
    """End-to-end: create_app() wired to a flex-auth double that allows

    everything -- proves the seam actually gates through create_app's own
    authorizer selection, not just when constructed directly.
    """

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})

    client = FlexAuthCheckClient(
        base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
    )
    app = create_app(authorizer=FlexAuthWriteAuthorizer(client=client))

    from fastapi.testclient import TestClient

    test_client = TestClient(app)
    created = test_client.post(
        "/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
    )
    assert created.status_code == 201