tenant-engine/tests/test_api_writes.py
tegwick 749461b97b
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 41s
Implement PostgreSQL production store path
Add the PostgreSQL backend, migration and stopped-write transfer tools, lease-aware deployment manifests, tenancy declarations, and shared conformance coverage. Persist grouping mutations in durable stores and separate process liveness from database readiness.
2026-08-19 14:43:08 +02:00

140 lines
4.5 KiB
Python

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", params={"actor": "tenant-engine"})
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", params={"actor": "tenant-engine"})
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