qonto-assistant/tests/test_live_authorization_gate.py

from pathlib import Path

import httpx
import pytest

from qonto_assistant.audit import AuditLogger
from qonto_assistant.contracts import ActorClaims
from qonto_assistant.errors import PolicyDeniedError
from qonto_assistant.flex_auth_client import FlexAuthCheckClient
from qonto_assistant.live_authorization import DENY_LIVE_AUTHZ, DENY_TENANT_ROLE, LiveAuthorizationGate
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.service import CapabilityService
from qonto_assistant.tenant_engine_client import TenantEngineClient

POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"


def _claims(tenant_id: str = "tenant:friendly:binky") -> ActorClaims:
    return ActorClaims(actor_id="agent-harness-binky", tenant_id=tenant_id, lane="green")


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

    return FlexAuthCheckClient(base_url="https://flex-auth.test", transport=httpx.MockTransport(handler))


def _tenant_engine_client(roles: list[str]) -> TenantEngineClient:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": roles})

    return TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler))


# --- Gate unit tests -------------------------------------------------------------------


def test_gate_allows_when_flex_auth_allows_and_role_matches() -> None:
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=_tenant_engine_client(["VEN"]),
        required_tenant_roles=frozenset({"VEN", "CUS"}),
    )

    assert gate.check(_claims()) is None


def test_gate_denies_when_flex_auth_denies() -> None:
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("deny"),
        tenant_engine_client=_tenant_engine_client(["VEN"]),
        required_tenant_roles=frozenset({"VEN", "CUS"}),
    )

    assert gate.check(_claims()) == DENY_LIVE_AUTHZ


def test_gate_denies_when_tenant_lacks_required_role() -> None:
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=_tenant_engine_client(["PLTF"]),  # not VEN/CUS
        required_tenant_roles=frozenset({"VEN", "CUS"}),
    )

    assert gate.check(_claims()) == DENY_TENANT_ROLE


def test_gate_skips_tenant_role_check_when_no_roles_required() -> None:
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=_tenant_engine_client([]),
        required_tenant_roles=frozenset(),
    )

    assert gate.check(_claims()) is None


def test_gate_skips_tenant_role_check_when_no_tenant_engine_configured() -> None:
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=None,
        required_tenant_roles=frozenset({"VEN"}),
    )

    assert gate.check(_claims()) is None


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

    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler)),
        required_tenant_roles=frozenset({"VEN"}),
    )

    assert gate.check(_claims()) == DENY_TENANT_ROLE


# --- CapabilityService integration -----------------------------------------------------


def _service(gate: LiveAuthorizationGate, events: list[dict[str, object]]) -> CapabilityService:
    return CapabilityService(
        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
        policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False),
        audit_logger=AuditLogger(sink=events.append),
        rate_limiter=RateLimiter(limit=20, window_seconds=60),
        concurrency_limiter=ConcurrencyLimiter(limit=4),
        live_authorization_gate=gate,
    )


def test_service_denies_capability_call_when_gate_denies() -> None:
    events: list[dict[str, object]] = []
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("deny"),
        tenant_engine_client=None,
        required_tenant_roles=frozenset(),
    )
    service = _service(gate, events)

    with pytest.raises(PolicyDeniedError) as exc_info:
        service.get_accounts(claims=_claims(), request_id="req-1", protocol="rest")

    assert exc_info.value.decision.reason == DENY_LIVE_AUTHZ
    deny_events = [e for e in events if e.get("deny_reason") == DENY_LIVE_AUTHZ]
    assert len(deny_events) == 1


def test_service_allows_capability_call_when_gate_allows() -> None:
    events: list[dict[str, object]] = []
    gate = LiveAuthorizationGate(
        flex_auth_client=_flex_auth_client("allow"),
        tenant_engine_client=_tenant_engine_client(["VEN"]),
        required_tenant_roles=frozenset({"VEN", "CUS"}),
    )
    service = _service(gate, events)

    payload = service.get_accounts(claims=_claims(), request_id="req-2", protocol="rest")

    assert "organization" in payload


def test_service_without_gate_configured_behaves_as_before() -> None:
    events: list[dict[str, object]] = []
    service = CapabilityService(
        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
        policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False),
        audit_logger=AuditLogger(sink=events.append),
        rate_limiter=RateLimiter(limit=20, window_seconds=60),
        concurrency_limiter=ConcurrencyLimiter(limit=4),
    )

    payload = service.get_accounts(claims=_claims(), request_id="req-3", protocol="rest")

    assert "organization" in payload