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
271 lines
9.8 KiB
Python
271 lines
9.8 KiB
Python
"""TEN-WP-0006-T03: one guardrail persistence contract, both store backends.
|
|
|
|
Parametrised over the in-memory and SQLite stores for the same reason the
|
|
lifecycle conformance suite is: the divergences that matter here -- atomicity
|
|
of override + audit + receipt, and the retired-tenant guard -- are 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 Tenant, TenantRetiredError
|
|
from tenant_engine.guardrail import (
|
|
UNLIMITED,
|
|
LimitKind,
|
|
LimitValue,
|
|
Provenance,
|
|
UnknownLimitKeyError,
|
|
resolve_limit,
|
|
)
|
|
from tenant_engine.sqlite_store import SQLiteTenantStore
|
|
from tenant_engine.store import (
|
|
IdempotencyConflictError,
|
|
InMemoryTenantStore,
|
|
VersionConflictError,
|
|
)
|
|
|
|
NOW = datetime(2026, 8, 16, 12, 0, tzinfo=UTC)
|
|
KEY = "spend.monthly"
|
|
|
|
|
|
@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:
|
|
t = Tenant.create(tenant_id="t-1", identifier="tenant:small:acme", created_at=NOW)
|
|
store.create_tenant(t)
|
|
return t
|
|
|
|
|
|
def eur(amount) -> LimitValue:
|
|
return LimitValue(kind=LimitKind.SPEND, amount=amount, currency="EUR", period="P1M")
|
|
|
|
|
|
def set_override(store, tenant, value, *, version=1, key=KEY, **kwargs):
|
|
params = dict(
|
|
tenant_id=tenant.tenant_id,
|
|
expected_version=version,
|
|
limit_key=key,
|
|
value=value,
|
|
change_id="c-1",
|
|
changed_by="ops",
|
|
reason="test",
|
|
correlation_id="corr-1",
|
|
idempotency_key="idem-1",
|
|
request_fingerprint="fp-1",
|
|
at=NOW,
|
|
)
|
|
params.update(kwargs)
|
|
return store.set_guardrail_override(**params)
|
|
|
|
|
|
# --- No backfill needed ---------------------------------------------------
|
|
|
|
|
|
def test_a_tenant_with_no_override_resolves_to_its_grouping_default(store, tenant):
|
|
# Grouping defaults are resolved, never materialised into rows. That is
|
|
# why this workplan needs no backfill migration: a tenant created before
|
|
# guardrails existed and one created after take the identical path, so an
|
|
# existing tenant cannot silently end up looser than a fresh one.
|
|
assert store.guardrail_overrides(tenant.tenant_id) == {}
|
|
effective = resolve_limit(KEY, tenant=store.get_tenant(tenant.tenant_id))
|
|
assert effective.provenance is Provenance.GROUPING
|
|
|
|
|
|
def test_the_migration_is_idempotent(tmp_path):
|
|
path = str(tmp_path / "tenant.db")
|
|
first = SQLiteTenantStore(path)
|
|
t = Tenant.create(tenant_id="t-1", identifier="tenant:small:acme", created_at=NOW)
|
|
first.create_tenant(t)
|
|
set_override(first, t, eur(9_000))
|
|
|
|
# forward-only and idempotent: reopening runs the same schema script
|
|
second = SQLiteTenantStore(path)
|
|
assert second.guardrail_overrides("t-1")[KEY].amount == 9_000
|
|
assert second.get_tenant("t-1").version == 2
|
|
|
|
|
|
# --- Write, read back, audit ---------------------------------------------
|
|
|
|
|
|
def test_setting_an_override_persists_it_and_bumps_the_version(store, tenant):
|
|
updated, change, replayed = set_override(store, tenant, eur(9_000))
|
|
assert not replayed
|
|
assert updated.version == 2
|
|
assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 9_000
|
|
assert change.previous is None
|
|
assert change.current.amount == 9_000
|
|
|
|
|
|
def test_the_override_wins_over_the_grouping_default(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
effective = resolve_limit(
|
|
KEY,
|
|
tenant=store.get_tenant(tenant.tenant_id),
|
|
overrides=store.guardrail_overrides(tenant.tenant_id),
|
|
)
|
|
assert effective.provenance is Provenance.OVERRIDE
|
|
assert effective.value.amount == 9_000
|
|
|
|
|
|
def test_the_change_is_audited_with_actor_reason_and_correlation(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
changes = store.guardrail_changes(tenant.tenant_id)
|
|
assert len(changes) == 1
|
|
assert (changes[0].changed_by, changes[0].reason, changes[0].correlation_id) == (
|
|
"ops",
|
|
"test",
|
|
"corr-1",
|
|
)
|
|
|
|
|
|
def test_the_audit_trail_is_append_only(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
set_override(store, tenant, eur(1_000), version=2, change_id="c-2", idempotency_key="idem-2")
|
|
changes = store.guardrail_changes(tenant.tenant_id)
|
|
assert [c.change_id for c in changes] == ["c-1", "c-2"]
|
|
# the earlier value survives in the record, not just the current one
|
|
assert changes[1].previous.amount == 9_000
|
|
|
|
|
|
def test_a_guardrail_change_emits_a_domain_event(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
events = [e for e in store.events_for(tenant.tenant_id) if e.event_type == "guardrail_changed"]
|
|
assert len(events) == 1
|
|
assert events[0].payload["limit_key"] == KEY
|
|
assert events[0].payload["correlation_id"] == "corr-1"
|
|
|
|
|
|
def test_clearing_an_override_falls_back_to_the_grouping_default(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
_, change, _ = set_override(
|
|
store, tenant, None, version=2, change_id="c-2", idempotency_key="idem-2"
|
|
)
|
|
assert change.is_clear
|
|
assert store.guardrail_overrides(tenant.tenant_id) == {}
|
|
effective = resolve_limit(KEY, tenant=store.get_tenant(tenant.tenant_id))
|
|
assert effective.provenance is Provenance.GROUPING
|
|
|
|
|
|
def test_unlimited_survives_a_round_trip_as_an_explicit_value(store, tenant):
|
|
entity = LimitValue(kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED)
|
|
# spend.monthly is the only registered key, so use it to prove the
|
|
# sentinel serialises; the kind check lives in the domain tests
|
|
set_override(
|
|
store,
|
|
tenant,
|
|
LimitValue(kind=LimitKind.SPEND, amount=UNLIMITED, currency="EUR", period="P1M"),
|
|
)
|
|
stored = store.guardrail_overrides(tenant.tenant_id)[KEY]
|
|
assert stored.is_unlimited
|
|
assert entity.is_unlimited
|
|
|
|
|
|
# --- Concurrency, idempotency, unknown keys ------------------------------
|
|
|
|
|
|
def test_a_stale_version_conflicts(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
with pytest.raises(VersionConflictError):
|
|
set_override(
|
|
store, tenant, eur(1_000), version=1, change_id="c-2", idempotency_key="idem-2"
|
|
)
|
|
|
|
|
|
def test_replay_returns_the_original_result_without_reapplying(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
updated, change, replayed = set_override(store, tenant, eur(9_000))
|
|
assert replayed
|
|
assert updated.version == 2 # not bumped a second time
|
|
assert change.change_id == "c-1"
|
|
assert len(store.guardrail_changes(tenant.tenant_id)) == 1
|
|
|
|
|
|
def test_reusing_a_key_for_a_different_request_conflicts(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
with pytest.raises(IdempotencyConflictError):
|
|
set_override(store, tenant, eur(1_000), version=2, request_fingerprint="fp-2")
|
|
|
|
|
|
def test_an_unregistered_key_is_rejected_before_anything_is_written(store, tenant):
|
|
with pytest.raises(UnknownLimitKeyError):
|
|
set_override(store, tenant, eur(9_000), key="spend.weekly")
|
|
assert store.get_tenant(tenant.tenant_id).version == 1
|
|
assert store.guardrail_changes(tenant.tenant_id) == []
|
|
|
|
|
|
def test_a_failed_write_leaves_no_audit_record_and_no_version_bump(store, tenant):
|
|
set_override(store, tenant, eur(9_000))
|
|
with pytest.raises(VersionConflictError):
|
|
set_override(store, tenant, eur(1), version=99, change_id="c-2", idempotency_key="idem-2")
|
|
assert store.get_tenant(tenant.tenant_id).version == 2
|
|
assert len(store.guardrail_changes(tenant.tenant_id)) == 1
|
|
assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 9_000
|
|
|
|
|
|
# --- Retired tenants ------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def retired(store, tenant) -> Tenant:
|
|
updated, _ = store.mutate_tenant(
|
|
tenant_id=tenant.tenant_id,
|
|
expected_version=1,
|
|
mutate=lambda t: t.retire(at=NOW),
|
|
event_type="tenant_retired",
|
|
evidence={"reason": "test"},
|
|
idempotency_key="retire-1",
|
|
request_fingerprint="fp-retire",
|
|
)
|
|
return updated
|
|
|
|
|
|
def test_a_retired_tenants_limits_clamp_to_the_floor_but_stay_readable(store, retired):
|
|
effective = resolve_limit(
|
|
KEY,
|
|
tenant=store.get_tenant(retired.tenant_id),
|
|
overrides=store.guardrail_overrides(retired.tenant_id),
|
|
)
|
|
assert effective.provenance is Provenance.LIFECYCLE
|
|
assert effective.value.amount == 0
|
|
|
|
|
|
def test_loosening_a_retired_tenants_guardrail_is_refused(store, retired):
|
|
# small's default is 25_000; 90_000 would take effect on reactivation
|
|
with pytest.raises(TenantRetiredError):
|
|
set_override(store, retired, eur(90_000), version=2)
|
|
assert store.guardrail_overrides(retired.tenant_id) == {}
|
|
|
|
|
|
def test_tightening_a_retired_tenants_guardrail_is_allowed(store, retired):
|
|
# reduce-privilege operations stay available while retired -- refusing
|
|
# them would be the fail-open choice
|
|
_, change, _ = set_override(store, retired, eur(100), version=2)
|
|
assert change.current.amount == 100
|
|
|
|
|
|
def test_clearing_an_override_that_would_loosen_is_refused_while_retired(store, tenant):
|
|
set_override(store, tenant, eur(100))
|
|
store.mutate_tenant(
|
|
tenant_id=tenant.tenant_id,
|
|
expected_version=2,
|
|
mutate=lambda t: t.retire(at=NOW),
|
|
event_type="tenant_retired",
|
|
evidence={},
|
|
idempotency_key="retire-1",
|
|
request_fingerprint="fp-retire",
|
|
)
|
|
# clearing would fall back to small's 25_000 default -- a loosening
|
|
with pytest.raises(TenantRetiredError):
|
|
set_override(store, tenant, None, version=3, change_id="c-3", idempotency_key="idem-3")
|
|
assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 100
|