tenant-engine/tests/test_flex_auth.py

from pathlib import Path

import httpx
import pytest

from tenant_engine.flex_auth import CheckRequest, FlexAuthCheckClient, new_request_id


def _request() -> CheckRequest:
    return CheckRequest(
        request_id=new_request_id(),
        tenant="tenant:friendly:binky",
        subject_id="tenant-engine",
        subject_type="service",
        action="tenant.create",
        resource_id="t-1",
        resource_type="tenant",
    )


def _client(handler) -> FlexAuthCheckClient:
    return FlexAuthCheckClient(
        base_url="https://flex-auth.example.test",
        timeout_seconds=1,
        transport=httpx.MockTransport(handler),
    )


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

    assert _client(handler).is_allowed(_request()) is True


@pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"])
def test_non_allow_effects_deny(effect: str) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}},
        )

    assert _client(handler).is_allowed(_request()) is False


def test_non_200_status_denies() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, json={"error": "internal"})

    assert _client(handler).is_allowed(_request()) is False


def test_malformed_json_body_denies() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, content=b"not json")

    assert _client(handler).is_allowed(_request()) is False


def test_non_object_json_body_denies() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=["not", "an", "object"])

    assert _client(handler).is_allowed(_request()) is False


def test_connection_failure_denies() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused", request=request)

    assert _client(handler).is_allowed(_request()) is False


def test_timeout_denies() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.TimeoutException("timed out", request=request)

    assert _client(handler).is_allowed(_request()) is False


def test_request_body_matches_schema_shape() -> None:
    seen: dict[str, object] = {}

    def handler(request: httpx.Request) -> httpx.Response:
        import json

        seen.update(json.loads(request.content))
        return httpx.Response(
            200,
            json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}},
        )

    _client(handler).is_allowed(_request())

    assert seen["tenant"] == "tenant:friendly:binky"
    assert seen["action"] == "tenant.create"
    assert seen["subject"] == {"id": "tenant-engine", "type": "service"}
    assert seen["resource"] == {"id": "t-1", "type": "tenant", "system": "tenant-engine"}


def test_rotating_caller_token_is_read_for_each_check(tmp_path: Path) -> None:
    token_file = tmp_path / "token"
    token_file.write_text("token-one\n")
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request.headers["authorization"])
        return httpx.Response(200, json={"id": "d-1", "effect": "allow"})

    client = FlexAuthCheckClient(
        base_url="https://flex-auth.example.test",
        transport=httpx.MockTransport(handler),
        bearer_token_file=str(token_file),
    )
    assert client.is_allowed(_request()) is True
    token_file.write_text("token-two\n")
    assert client.is_allowed(_request()) is True
    assert seen == ["Bearer token-one", "Bearer token-two"]