tegwick
adb74d2443
- authz.py: WriteAuthorizer Protocol + DefaultDenyWriteAuthorizer. Every
write endpoint calls it before touching the store; denial maps to
403 write_denied via an exception handler.
- app.py: GET /tenants/{id}/roles (cache-read, key-cape) and
GET /tenants/{id}/roles/live (live-lookup, flex-auth) share one handler
that fails closed (503) on StoreUnavailableError -- deliberately made
identical rather than giving cache-read weaker guarantees than the task
strictly required. POST /tenants, /roles/grant, /roles/revoke, /plan --
all four gated by the WriteAuthorizer seam, domain/store errors mapped to
400/404/409 after authorization passes.
- store.py: new StoreUnavailableError for the fail-closed test double.
43 tests passing: default-deny on every write endpoint, an
_AllowAllAuthorizer test double proving the seam actually gates (full
create->grant->read->revoke->read->assign-plan lifecycle over real HTTP),
and a _BrokenStore double proving outage never looks like "zero roles".
Verified live over real HTTP, not just TestClient.
TEN-WP-0002 closed: all 7 tasks done, boundary-contract ownership checked
against the implementation with no drift found. Follow-ups recorded in the
closure note (real flex-auth WriteAuthorizer, key-cape wiring, guardrail
policy design, Binky as first real tenant record, durable persistence).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
100 lines
3 KiB
Python
100 lines
3 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from tenant_engine.app import create_app
|
|
from tenant_engine.domain import CapabilityRole, Tenant, create_role_grant
|
|
from tenant_engine.store import InMemoryTenantStore, TenantStore
|
|
|
|
|
|
class _BrokenStore:
|
|
"""Test double: every active_roles() call raises, simulating an outage."""
|
|
|
|
def __init__(self, delegate: TenantStore) -> None:
|
|
self._delegate = delegate
|
|
|
|
def create_tenant(self, tenant):
|
|
return self._delegate.create_tenant(tenant)
|
|
|
|
def get_tenant(self, tenant_id):
|
|
return self._delegate.get_tenant(tenant_id)
|
|
|
|
def grant_role(self, grant):
|
|
return self._delegate.grant_role(grant)
|
|
|
|
def revoke_role(self, **kwargs):
|
|
return self._delegate.revoke_role(**kwargs)
|
|
|
|
def active_roles(self, tenant_id):
|
|
from tenant_engine.store import StoreUnavailableError
|
|
|
|
raise StoreUnavailableError("simulated outage")
|
|
|
|
def assign_plan(self, assignment):
|
|
return self._delegate.assign_plan(assignment)
|
|
|
|
def events(self):
|
|
return self._delegate.events()
|
|
|
|
|
|
def _seeded_store() -> InMemoryTenantStore:
|
|
store = InMemoryTenantStore()
|
|
tenant = Tenant.create(tenant_id="t-binky", identifier="tenant:friendly:binky")
|
|
store.create_tenant(tenant)
|
|
store.grant_role(
|
|
create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.CUS,
|
|
grant_reason="manual_grant",
|
|
plan_id=None,
|
|
granted_by="ops",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
return store
|
|
|
|
|
|
def test_cache_read_roles_returns_active_roles() -> None:
|
|
client = TestClient(create_app(store=_seeded_store()))
|
|
response = client.get("/tenants/t-binky/roles")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"tenant_id": "t-binky", "roles": ["CUS"]}
|
|
|
|
|
|
def test_cache_read_roles_unknown_tenant_is_404() -> None:
|
|
client = TestClient(create_app(store=_seeded_store()))
|
|
response = client.get("/tenants/does-not-exist/roles")
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_live_lookup_roles_returns_active_roles() -> None:
|
|
client = TestClient(create_app(store=_seeded_store()))
|
|
response = client.get("/tenants/t-binky/roles/live")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["roles"] == ["CUS"]
|
|
|
|
|
|
def test_live_lookup_fails_closed_on_store_outage() -> None:
|
|
broken = _BrokenStore(_seeded_store())
|
|
client = TestClient(create_app(store=broken))
|
|
|
|
response = client.get("/tenants/t-binky/roles/live")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() != {"tenant_id": "t-binky", "roles": []}, (
|
|
"outage must not be indistinguishable from a legitimate empty role list"
|
|
)
|
|
|
|
|
|
def test_cache_read_also_fails_closed_on_store_outage() -> None:
|
|
broken = _BrokenStore(_seeded_store())
|
|
client = TestClient(create_app(store=broken))
|
|
|
|
response = client.get("/tenants/t-binky/roles")
|
|
|
|
assert response.status_code == 503
|