tegwick
0770ce82d9
Python 3.12 + FastAPI, pyproject.toml + Makefile mirroring
qonto-assistant's exactly. src/tenant_engine/{domain,store,app,main}.py:
- domain.py: Tenant, CapabilityRole (PLTF/IAM/VEN/CUS), RoleGrant,
PlanAssignment, create_role_grant() enforcing ADR-0014's invariants.
Refinement made while implementing: platform_default grants are valid
for trial-grouped tenants OR the reserved tenant:platform/tenant:coulomb
tenants (their baseline roles were never purchased either) -- the task
spec only named the trial case.
- store.py: TenantStore Protocol + InMemoryTenantStore, every mutation
emits a DomainEvent per the boundary contract's Audit Correlation
Contract.
- app.py/main.py: FastAPI factory + /health endpoint, verified live on
127.0.0.1:8090.
29 tests passing, including non-exclusive role coexistence (CUS+VEN
simultaneously) and append-only revoke semantics.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
124 lines
4 KiB
Python
124 lines
4 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from tenant_engine.domain import CapabilityRole, PlanAssignment, Tenant, create_role_grant
|
|
from tenant_engine.store import (
|
|
GrantNotFoundError,
|
|
InMemoryTenantStore,
|
|
TenantAlreadyExistsError,
|
|
TenantNotFoundError,
|
|
)
|
|
|
|
|
|
def _store_with_tenant(*, grouping: str = "friendly", name: str = "binky") -> tuple[InMemoryTenantStore, Tenant]:
|
|
store = InMemoryTenantStore()
|
|
tenant = Tenant.create(tenant_id=f"t-{name}", identifier=f"tenant:{grouping}:{name}")
|
|
store.create_tenant(tenant)
|
|
return store, tenant
|
|
|
|
|
|
def test_create_and_get_tenant() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
assert store.get_tenant(tenant.tenant_id) == tenant
|
|
|
|
|
|
def test_create_duplicate_tenant_raises() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
with pytest.raises(TenantAlreadyExistsError):
|
|
store.create_tenant(tenant)
|
|
|
|
|
|
def test_get_unknown_tenant_raises() -> None:
|
|
store = InMemoryTenantStore()
|
|
with pytest.raises(TenantNotFoundError):
|
|
store.get_tenant("does-not-exist")
|
|
|
|
|
|
def test_grant_and_active_roles() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
grant = 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),
|
|
)
|
|
store.grant_role(grant)
|
|
|
|
assert store.active_roles(tenant.tenant_id) == frozenset({CapabilityRole.CUS})
|
|
|
|
|
|
def test_revoke_removes_role_from_active_set_but_keeps_record() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
grant = create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.VEN,
|
|
grant_reason="manual_grant",
|
|
plan_id=None,
|
|
granted_by="ops",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
store.grant_role(grant)
|
|
revoked = store.revoke_role(tenant_id=tenant.tenant_id, grant_id="g-1", at=datetime.now(UTC))
|
|
|
|
assert revoked.revoked_at is not None
|
|
assert store.active_roles(tenant.tenant_id) == frozenset()
|
|
|
|
|
|
def test_revoke_unknown_grant_raises() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
with pytest.raises(GrantNotFoundError):
|
|
store.revoke_role(tenant_id=tenant.tenant_id, grant_id="missing", at=datetime.now(UTC))
|
|
|
|
|
|
def test_non_exclusive_roles_coexist() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
for role in (CapabilityRole.CUS, CapabilityRole.VEN):
|
|
store.grant_role(
|
|
create_role_grant(
|
|
tenant=tenant,
|
|
grant_id=f"g-{role.value}",
|
|
role=role,
|
|
grant_reason="manual_grant",
|
|
plan_id=None,
|
|
granted_by="ops",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
|
|
assert store.active_roles(tenant.tenant_id) == frozenset({CapabilityRole.CUS, CapabilityRole.VEN})
|
|
|
|
|
|
def test_assign_plan() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
store.assign_plan(PlanAssignment(tenant_id=tenant.tenant_id, plan_id="plan-x", assigned_at=datetime.now(UTC)))
|
|
|
|
events = store.events()
|
|
assert any(event.event_type == "plan_assigned" and event.payload["plan_id"] == "plan-x" for event in events)
|
|
|
|
|
|
def test_every_mutation_emits_an_event() -> None:
|
|
store, tenant = _store_with_tenant()
|
|
grant = 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),
|
|
)
|
|
store.grant_role(grant)
|
|
store.revoke_role(tenant_id=tenant.tenant_id, grant_id="g-1", at=datetime.now(UTC))
|
|
store.assign_plan(PlanAssignment(tenant_id=tenant.tenant_id, plan_id="plan-x", assigned_at=datetime.now(UTC)))
|
|
|
|
event_types = [event.event_type for event in store.events()]
|
|
assert event_types == ["tenant_created", "role_granted", "role_revoked", "plan_assigned"]
|