tegwick
672cf4da6e
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 37s
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
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""TEN-WP-0005-T02/T04: one lifecycle contract, every store backend.
|
|
|
|
Every test here is parametrised over the in-memory, SQLite, and PostgreSQL stores so the
|
|
durable backend cannot silently diverge from the reference semantics -- the
|
|
divergence that matters (CAS, idempotency, retirement guards) is exactly the
|
|
kind a single-backend suite would miss.
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from postgres_backend import clean_postgres_store
|
|
|
|
from tenant_engine.domain import (
|
|
CapabilityRole,
|
|
InvalidLifecycleTransitionError,
|
|
PlanAssignment,
|
|
Tenant,
|
|
TenantLifecycle,
|
|
TenantRetiredError,
|
|
create_role_grant,
|
|
)
|
|
from tenant_engine.sqlite_store import SQLiteTenantStore
|
|
from tenant_engine.store import (
|
|
IdempotencyConflictError,
|
|
InMemoryTenantStore,
|
|
TenantNotFoundError,
|
|
VersionConflictError,
|
|
)
|
|
|
|
NOW = datetime(2026, 8, 10, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
@pytest.fixture(params=["memory", "sqlite", "postgres"])
|
|
def store(request, tmp_path):
|
|
if request.param == "memory":
|
|
return InMemoryTenantStore()
|
|
if request.param == "sqlite":
|
|
return SQLiteTenantStore(str(tmp_path / "tenant.db"))
|
|
return clean_postgres_store(tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def tenant(store) -> Tenant:
|
|
record = Tenant.create(
|
|
tenant_id="t-1", identifier="tenant:friendly:binky", display_name="Binky", created_at=NOW
|
|
)
|
|
store.create_tenant(record)
|
|
return record
|
|
|
|
|
|
def _retire(store, *, key: str = "idem-retire", version: int = 1):
|
|
return store.mutate_tenant(
|
|
tenant_id="t-1",
|
|
expected_version=version,
|
|
mutate=lambda t: t.retire(at=NOW),
|
|
event_type="tenant_retired",
|
|
evidence={"actor": "ops", "reason": "offboarded", "correlation_id": "corr-1"},
|
|
idempotency_key=key,
|
|
request_fingerprint="fp-retire",
|
|
)
|
|
|
|
|
|
def _rename(store, *, key: str, version: int, name: str = "Binky Ltd", fingerprint: str = "fp-a"):
|
|
return store.mutate_tenant(
|
|
tenant_id="t-1",
|
|
expected_version=version,
|
|
mutate=lambda t: t.with_metadata({"display_name": name}, at=NOW),
|
|
event_type="tenant_updated",
|
|
evidence={"actor": "ops", "reason": "rename", "correlation_id": "corr-1"},
|
|
idempotency_key=key,
|
|
request_fingerprint=fingerprint,
|
|
)
|
|
|
|
|
|
def test_update_persists_and_bumps_version(store, tenant) -> None:
|
|
updated, replayed = _rename(store, key="k1", version=1)
|
|
|
|
assert replayed is False
|
|
assert updated.version == 2
|
|
assert store.get_tenant("t-1").display_name == "Binky Ltd"
|
|
|
|
|
|
def test_grouping_mutation_persists(store, tenant) -> None:
|
|
updated, replayed = store.mutate_tenant(
|
|
tenant_id="t-1",
|
|
expected_version=1,
|
|
mutate=lambda current: current.with_grouping("large", at=NOW),
|
|
event_type="tenant_grouping_changed",
|
|
evidence={"actor": "ops", "reason": "growth", "correlation_id": "corr-grouping"},
|
|
idempotency_key="grouping-large",
|
|
request_fingerprint="fp-grouping-large",
|
|
)
|
|
|
|
assert replayed is False
|
|
assert updated.grouping == "large"
|
|
assert store.get_tenant("t-1").grouping == "large"
|
|
|
|
|
|
def test_stale_version_is_rejected(store, tenant) -> None:
|
|
_rename(store, key="k1", version=1)
|
|
|
|
with pytest.raises(VersionConflictError) as exc:
|
|
_rename(store, key="k2", version=1, name="Binky GmbH")
|
|
|
|
assert exc.value.actual == 2
|
|
assert store.get_tenant("t-1").display_name == "Binky Ltd"
|
|
|
|
|
|
def test_duplicate_idempotency_key_replays_the_original_result(store, tenant) -> None:
|
|
first, _ = _rename(store, key="k1", version=1)
|
|
replay, replayed = _rename(store, key="k1", version=1)
|
|
|
|
assert replayed is True
|
|
assert replay == first
|
|
# The replay must not apply the mutation a second time.
|
|
assert store.get_tenant("t-1").version == 2
|
|
|
|
|
|
def test_conflicting_idempotency_key_reuse_is_rejected(store, tenant) -> None:
|
|
_rename(store, key="k1", version=1, fingerprint="fp-a")
|
|
|
|
with pytest.raises(IdempotencyConflictError):
|
|
_rename(store, key="k1", version=1, name="Something Else", fingerprint="fp-b")
|
|
|
|
|
|
def test_mutating_an_unknown_tenant_raises(store) -> None:
|
|
with pytest.raises(TenantNotFoundError):
|
|
_rename(store, key="k1", version=1)
|
|
|
|
|
|
def test_retire_then_reactivate_round_trip(store, tenant) -> None:
|
|
retired, _ = _retire(store)
|
|
assert retired.lifecycle is TenantLifecycle.RETIRED
|
|
assert store.get_tenant("t-1").lifecycle is TenantLifecycle.RETIRED
|
|
|
|
reactivated, _ = store.mutate_tenant(
|
|
tenant_id="t-1",
|
|
expected_version=2,
|
|
mutate=lambda t: t.reactivate(at=NOW),
|
|
event_type="tenant_reactivated",
|
|
evidence={"actor": "ops", "reason": "returned", "correlation_id": "corr-2"},
|
|
idempotency_key="idem-reactivate",
|
|
request_fingerprint="fp-reactivate",
|
|
)
|
|
assert reactivated.lifecycle is TenantLifecycle.ACTIVE
|
|
assert store.get_tenant("t-1").version == 3
|
|
|
|
|
|
def test_double_retirement_with_a_new_key_is_an_invalid_transition(store, tenant) -> None:
|
|
_retire(store)
|
|
|
|
with pytest.raises(InvalidLifecycleTransitionError):
|
|
_retire(store, key="idem-retire-2", version=2)
|
|
|
|
|
|
def test_failed_mutation_leaves_no_version_bump_and_no_receipt(store, tenant) -> None:
|
|
with pytest.raises(InvalidLifecycleTransitionError):
|
|
store.mutate_tenant(
|
|
tenant_id="t-1",
|
|
expected_version=1,
|
|
mutate=lambda t: t.reactivate(at=NOW), # already active
|
|
event_type="tenant_reactivated",
|
|
evidence={"actor": "ops", "reason": "x", "correlation_id": "c"},
|
|
idempotency_key="k-fail",
|
|
request_fingerprint="fp",
|
|
)
|
|
|
|
assert store.get_tenant("t-1").version == 1
|
|
# The failed key must be reusable -- a rolled-back attempt is not a receipt.
|
|
updated, replayed = _rename(store, key="k-fail", version=1)
|
|
assert replayed is False
|
|
assert updated.version == 2
|
|
|
|
|
|
def test_retired_tenant_refuses_new_grants_and_plan_changes(store, tenant) -> None:
|
|
_retire(store)
|
|
|
|
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=NOW,
|
|
)
|
|
with pytest.raises(TenantRetiredError):
|
|
store.grant_role(grant)
|
|
with pytest.raises(TenantRetiredError):
|
|
store.assign_plan(PlanAssignment(tenant_id="t-1", plan_id="plan-x", assigned_at=NOW))
|
|
|
|
|
|
def test_retirement_preserves_existing_grant_and_plan_history(store, tenant) -> None:
|
|
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=NOW,
|
|
)
|
|
)
|
|
store.assign_plan(PlanAssignment(tenant_id="t-1", plan_id="plan-x", assigned_at=NOW))
|
|
|
|
_retire(store)
|
|
|
|
# Retirement is not a revocation: history stays queryable for audit and
|
|
# so reactivation does not have to reconstruct anything.
|
|
assert store.active_roles("t-1") == frozenset({CapabilityRole.CUS})
|
|
assert any(event.event_type == "plan_assigned" for event in store.events_for(tenant.tenant_id))
|
|
|
|
|
|
def test_mutation_emits_a_correlated_audit_event(store, tenant) -> None:
|
|
_rename(store, key="k1", version=1)
|
|
|
|
event = [e for e in store.events_for(tenant.tenant_id) if e.event_type == "tenant_updated"][-1]
|
|
assert event.payload["actor"] == "ops"
|
|
assert event.payload["reason"] == "rename"
|
|
assert event.payload["correlation_id"] == "corr-1"
|
|
assert event.payload["version"] == 2
|
|
|
|
|
|
def test_lifecycle_survives_reopening_the_database(tmp_path) -> None:
|
|
path = str(tmp_path / "tenant.db")
|
|
store = SQLiteTenantStore(path)
|
|
store.create_tenant(
|
|
Tenant.create(tenant_id="t-1", identifier="tenant:friendly:binky", created_at=NOW)
|
|
)
|
|
_retire(store)
|
|
|
|
reopened = SQLiteTenantStore(path)
|
|
assert reopened.get_tenant("t-1").lifecycle is TenantLifecycle.RETIRED
|
|
assert reopened.get_tenant("t-1").version == 2
|
|
|
|
# Restart-safe idempotency: the receipt outlives the process.
|
|
replay, replayed = _retire(reopened, version=1)
|
|
assert replayed is True
|
|
assert replay.version == 2
|
|
|
|
|
|
def test_migration_defaults_pre_lifecycle_rows_to_active(tmp_path) -> None:
|
|
"""A database written by the pre-TEN-WP-0005 schema must open and read."""
|
|
import sqlite3
|
|
|
|
path = str(tmp_path / "legacy.db")
|
|
legacy = sqlite3.connect(path)
|
|
with legacy:
|
|
legacy.executescript("""
|
|
CREATE TABLE tenants (
|
|
tenant_id TEXT PRIMARY KEY, identifier TEXT UNIQUE NOT NULL, grouping_name TEXT
|
|
);
|
|
CREATE TABLE grants (
|
|
grant_id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, role TEXT NOT NULL,
|
|
grant_reason TEXT NOT NULL, plan_id TEXT, granted_by TEXT NOT NULL,
|
|
granted_at TEXT NOT NULL, correlation_id TEXT NOT NULL, revoked_at TEXT
|
|
);
|
|
CREATE TABLE plans (
|
|
tenant_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL, assigned_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE events (
|
|
seq INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL,
|
|
tenant_id TEXT NOT NULL, at TEXT NOT NULL, payload TEXT NOT NULL
|
|
);
|
|
INSERT INTO tenants VALUES ('t-legacy', 'tenant:friendly:legacy', 'friendly');
|
|
INSERT INTO grants VALUES ('g-legacy', 't-legacy', 'CUS', 'manual_grant', NULL,
|
|
'ops', '2026-08-01T00:00:00+00:00', 'corr-legacy', NULL);
|
|
INSERT INTO plans VALUES ('t-legacy', 'plan-legacy', '2026-08-01T00:00:00+00:00');
|
|
""")
|
|
legacy.close()
|
|
|
|
migrated = SQLiteTenantStore(path)
|
|
tenant = migrated.get_tenant("t-legacy")
|
|
|
|
assert tenant.lifecycle is TenantLifecycle.ACTIVE
|
|
assert tenant.version == 1
|
|
assert tenant.identifier == "tenant:friendly:legacy"
|
|
assert migrated.active_roles("t-legacy") == frozenset({CapabilityRole.CUS})
|
|
# And the migrated row is immediately mutable under the new contract.
|
|
updated, _ = migrated.mutate_tenant(
|
|
tenant_id="t-legacy",
|
|
expected_version=1,
|
|
mutate=lambda t: t.with_metadata({"display_name": "Legacy"}, at=NOW),
|
|
event_type="tenant_updated",
|
|
evidence={"actor": "ops", "reason": "backfill", "correlation_id": "c"},
|
|
idempotency_key="k",
|
|
request_fingerprint="fp",
|
|
)
|
|
assert updated.version == 2
|
|
|
|
|
|
def test_concurrent_writers_only_one_wins(tmp_path) -> None:
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
store = SQLiteTenantStore(str(tmp_path / "tenant.db"))
|
|
store.create_tenant(
|
|
Tenant.create(tenant_id="t-1", identifier="tenant:friendly:binky", created_at=NOW)
|
|
)
|
|
|
|
def attempt(n: int):
|
|
try:
|
|
return _rename(store, key=f"k{n}", version=1, name=f"Name {n}", fingerprint=f"fp{n}")[0]
|
|
except VersionConflictError:
|
|
return None
|
|
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
results = list(pool.map(attempt, range(8)))
|
|
|
|
winners = [r for r in results if r is not None]
|
|
assert len(winners) == 1
|
|
assert store.get_tenant("t-1").version == 2
|