tenant-engine/tests/test_api_writes.py
tegwick 672cf4da6e
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 37s
Implement TEN-WP-0011 security layer conformance
Engine/PIP declaration is now checkable (layer.yaml plus a Tooling-client
scan). Writes persist a decision record or the published fail-closed
stance, live-lookup freshness is published, events_for is tenant-scoped,
and mutation evidence drains to audit-core from a local outbox without
blocking the mutation.

Sender registration is requested as AUDIT-IN-0002. Boundary-contract
amendment is requested as NET-IN-0002.

Assistant: grok
Assistant-Session: 01a04cea-e5e8-7081-a0fc-808ebbc35fa9
2026-08-29 13:02:51 +02:00

139 lines
4.3 KiB
Python

from fastapi.testclient import TestClient
from helpers import AllowAllAuthorizer
from tenant_engine.app import create_app
from tenant_engine.store import InMemoryTenantStore
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