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>
206 lines
6 KiB
Python
206 lines
6 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from tenant_engine.domain import (
|
|
CapabilityRole,
|
|
InvalidGrantError,
|
|
InvalidTenantIdentifierError,
|
|
Tenant,
|
|
create_role_grant,
|
|
parse_tenant_identifier,
|
|
)
|
|
|
|
|
|
def test_parse_tenant_identifier_reserved_platform() -> None:
|
|
grouping, name = parse_tenant_identifier("tenant:platform")
|
|
assert grouping is None
|
|
assert name == "platform"
|
|
|
|
|
|
def test_parse_tenant_identifier_reserved_coulomb() -> None:
|
|
grouping, name = parse_tenant_identifier("tenant:coulomb")
|
|
assert grouping is None
|
|
assert name == "coulomb"
|
|
|
|
|
|
def test_parse_tenant_identifier_grouped() -> None:
|
|
grouping, name = parse_tenant_identifier("tenant:friendly:binky")
|
|
assert grouping == "friendly"
|
|
assert name == "binky"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"identifier",
|
|
[
|
|
"tenant:unknown-grouping:binky",
|
|
"tenant:friendly",
|
|
"tenant:friendly:",
|
|
"not-a-tenant:friendly:binky",
|
|
"friendly:binky",
|
|
],
|
|
)
|
|
def test_parse_tenant_identifier_rejects_invalid_shapes(identifier: str) -> None:
|
|
with pytest.raises(InvalidTenantIdentifierError):
|
|
parse_tenant_identifier(identifier)
|
|
|
|
|
|
def test_tenant_create_from_identifier() -> None:
|
|
tenant = Tenant.create(tenant_id="t-1", identifier="tenant:friendly:binky")
|
|
assert tenant.grouping == "friendly"
|
|
assert tenant.is_reserved is False
|
|
|
|
|
|
def test_tenant_create_reserved_has_no_grouping() -> None:
|
|
tenant = Tenant.create(tenant_id="t-platform", identifier="tenant:platform")
|
|
assert tenant.grouping is None
|
|
assert tenant.is_reserved is True
|
|
|
|
|
|
def _tenant(*, grouping: str | None, identifier: str | None = None) -> Tenant:
|
|
if identifier is None:
|
|
identifier = f"tenant:{grouping}:acme" if grouping else "tenant:platform"
|
|
return Tenant.create(tenant_id="t-1", identifier=identifier)
|
|
|
|
|
|
def test_plan_assignment_grant_requires_plan_id() -> None:
|
|
tenant = _tenant(grouping="friendly")
|
|
with pytest.raises(InvalidGrantError):
|
|
create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.IAM,
|
|
grant_reason="plan_assignment",
|
|
plan_id=None,
|
|
granted_by="ops",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
def test_plan_assignment_grant_with_plan_id_succeeds() -> None:
|
|
tenant = _tenant(grouping="friendly")
|
|
grant = create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.IAM,
|
|
grant_reason="plan_assignment",
|
|
plan_id="plan-iam-dedicated",
|
|
granted_by="ops",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
assert grant.active is True
|
|
assert grant.plan_id == "plan-iam-dedicated"
|
|
|
|
|
|
def test_platform_default_allowed_for_trial_tenant_without_plan() -> None:
|
|
tenant = _tenant(grouping="trial")
|
|
grant = create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.VEN,
|
|
grant_reason="platform_default",
|
|
plan_id=None,
|
|
granted_by="platform",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
assert grant.grant_reason == "platform_default"
|
|
assert grant.plan_id is None
|
|
|
|
|
|
def test_platform_default_allowed_for_reserved_tenant() -> None:
|
|
tenant = _tenant(grouping=None, identifier="tenant:platform")
|
|
grant = create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.PLTF,
|
|
grant_reason="platform_default",
|
|
plan_id=None,
|
|
granted_by="platform",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
assert grant.role is CapabilityRole.PLTF
|
|
|
|
|
|
def test_platform_default_rejected_for_non_trial_grouped_tenant() -> None:
|
|
tenant = _tenant(grouping="enterprise")
|
|
with pytest.raises(InvalidGrantError):
|
|
create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.VEN,
|
|
grant_reason="platform_default",
|
|
plan_id=None,
|
|
granted_by="platform",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
def test_platform_default_rejected_when_plan_id_present() -> None:
|
|
tenant = _tenant(grouping="trial")
|
|
with pytest.raises(InvalidGrantError):
|
|
create_role_grant(
|
|
tenant=tenant,
|
|
grant_id="g-1",
|
|
role=CapabilityRole.VEN,
|
|
grant_reason="platform_default",
|
|
plan_id="plan-x",
|
|
granted_by="platform",
|
|
correlation_id="corr-1",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
def test_manual_grant_allows_missing_plan_id_for_any_grouping() -> None:
|
|
tenant = _tenant(grouping="enterprise")
|
|
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),
|
|
)
|
|
assert grant.grant_reason == "manual_grant"
|
|
|
|
|
|
def test_grant_revoke_is_append_only() -> None:
|
|
tenant = _tenant(grouping="friendly")
|
|
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),
|
|
)
|
|
revoked = grant.revoke(at=datetime.now(UTC))
|
|
|
|
assert grant.revoked_at is None, "original record must be untouched"
|
|
assert revoked.revoked_at is not None
|
|
assert revoked.grant_id == grant.grant_id
|
|
|
|
|
|
def test_revoking_twice_raises() -> None:
|
|
tenant = _tenant(grouping="friendly")
|
|
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),
|
|
).revoke(at=datetime.now(UTC))
|
|
|
|
with pytest.raises(InvalidGrantError):
|
|
grant.revoke(at=datetime.now(UTC))
|