tenant-engine/tests/test_api_writes.py

from fastapi.testclient import TestClient

from tenant_engine.app import create_app
from tenant_engine.authz import WriteAuthorizer
from tenant_engine.store import InMemoryTenantStore


class _AllowAllAuthorizer(WriteAuthorizer):
    def authorize(self, *, action: str, tenant_id: str, actor: str) -> None:
        return None


def _client(*, allow: bool = False) -> TestClient:
    store = InMemoryTenantStore()
    authorizer = _AllowAllAuthorizer() if allow else None
    return TestClient(create_app(store=store, authorizer=authorizer))


def test_create_tenant_denied_by_default() -> None:
    client = _client()
    response = client.post(
        "/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
    )
    assert response.status_code == 403
    assert response.json()["error_code"] == "write_denied"


def test_grant_role_denied_by_default() -> None:
    client = _client()
    response = client.post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "corr-1",
            "actor": "ops",
        },
    )
    assert response.status_code == 403


def test_revoke_role_denied_by_default() -> None:
    client = _client()
    response = client.post("/tenants/t-1/roles/revoke", json={"grant_id": "g-1", "actor": "ops"})
    assert response.status_code == 403


def test_assign_plan_denied_by_default() -> None:
    client = _client()
    response = client.post("/tenants/t-1/plan", json={"plan_id": "plan-x", "actor": "ops"})
    assert response.status_code == 403


def test_full_write_lifecycle_succeeds_when_authorizer_allows() -> None:
    client = _client(allow=True)

    created = client.post(
        "/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
    )
    assert created.status_code == 201
    assert created.json()["grouping"] == "friendly"

    granted = client.post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "corr-1",
            "actor": "ops",
        },
    )
    assert granted.status_code == 201

    roles = client.get("/tenants/t-1/roles")
    assert roles.json()["roles"] == ["CUS"]

    revoked = client.post("/tenants/t-1/roles/revoke", json={"grant_id": "g-1", "actor": "ops"})
    assert revoked.status_code == 200

    roles_after = client.get("/tenants/t-1/roles")
    assert roles_after.json()["roles"] == []

    plan = client.post("/tenants/t-1/plan", json={"plan_id": "plan-x", "actor": "ops"})
    assert plan.status_code == 200
    assert plan.json()["plan_id"] == "plan-x"


def test_create_tenant_rejects_invalid_identifier_after_authorization() -> None:
    client = _client(allow=True)
    response = client.post(
        "/tenants", json={"tenant_id": "t-1", "identifier": "tenant:unknown:binky", "actor": "ops"}
    )
    assert response.status_code == 400


def test_create_tenant_duplicate_is_409() -> None:
    client = _client(allow=True)
    client.post("/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"})
    response = client.post(
        "/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
    )
    assert response.status_code == 409


def test_grant_role_plan_assignment_without_plan_id_is_400() -> None:
    client = _client(allow=True)
    client.post("/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"})

    response = client.post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "IAM",
            "grant_reason": "plan_assignment",
            "granted_by": "ops",
            "correlation_id": "corr-1",
            "actor": "ops",
        },
    )
    assert response.status_code == 400


def test_grant_role_unknown_tenant_is_404() -> None:
    client = _client(allow=True)
    response = client.post(
        "/tenants/does-not-exist/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "corr-1",
            "actor": "ops",
        },
    )
    assert response.status_code == 404