from datetime import UTC, datetime import pytest from tenant_engine.domain import GROUPINGS, RESERVED_IDENTIFIERS, Tenant from tenant_engine.guardrail import ( UNLIMITED, ConflictingLimitError, GuardrailChange, GuardrailRegistryInvalidError, InvalidLimitError, LimitDefinition, LimitKind, LimitRegistry, LimitValue, Provenance, UnknownLimitKeyError, resolve_limit, resolve_limits, ) from tenant_engine.guardrail.registry import ( DEFAULT_REGISTRY, SPEND_MONTHLY, unlimited_reserved_default, ) NOW = datetime(2026, 8, 16, 12, 0, tzinfo=UTC) def tenant(identifier: str, *, retired: bool = False) -> Tenant: t = Tenant.create(tenant_id="t-1", identifier=identifier, created_at=NOW) return t.retire(at=NOW) if retired else t def eur(amount) -> LimitValue: return LimitValue(kind=LimitKind.SPEND, amount=amount, currency="EUR", period="P1M") # --- LimitValue invariants ------------------------------------------------ def test_spend_limit_requires_currency_and_period(): with pytest.raises(InvalidLimitError): LimitValue(kind=LimitKind.SPEND, amount=100, period="P1M") with pytest.raises(InvalidLimitError): LimitValue(kind=LimitKind.SPEND, amount=100, currency="EUR") def test_count_limits_reject_currency_and_enforce_period_shape(): with pytest.raises(InvalidLimitError): LimitValue(kind=LimitKind.ENTITY_COUNT, amount=5, currency="EUR") with pytest.raises(InvalidLimitError): # an entity count is a point-in-time cap, not a rate LimitValue(kind=LimitKind.ENTITY_COUNT, amount=5, period="P1M") with pytest.raises(InvalidLimitError): LimitValue(kind=LimitKind.ACTION_COUNT, amount=5) def test_negative_and_bool_amounts_are_rejected(): with pytest.raises(InvalidLimitError): eur(-1) with pytest.raises(InvalidLimitError): eur(True) def test_unlimited_is_explicit_and_never_an_int(): value = LimitValue(kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED) assert value.is_unlimited assert not eur(0).is_unlimited def test_reduced_to_floor_only_ever_reduces(): floor = eur(0) assert eur(500).reduced_to_floor(floor) == floor assert eur(0).reduced_to_floor(eur(500)) == eur(0) assert LimitValue( kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED ).reduced_to_floor(LimitValue(kind=LimitKind.ENTITY_COUNT, amount=3)).amount == 3 def test_clamping_across_kinds_is_a_conflict(): with pytest.raises(ConflictingLimitError): eur(10).reduced_to_floor(LimitValue(kind=LimitKind.ENTITY_COUNT, amount=0)) def test_currency_is_not_part_of_compatibility(): # precedence resolves per key and the top layer wins outright, so two # currencies never combine -- tenant-engine holds no exchange rate usd = LimitValue(kind=LimitKind.SPEND, amount=100, currency="USD", period="P1M") assert eur(100).compatible_with(usd) # --- Registry ------------------------------------------------------------- def test_registry_requires_a_default_for_every_grouping(): partial = {g: eur(0) for g in list(GROUPINGS)[:-1]} with pytest.raises(GuardrailRegistryInvalidError): LimitDefinition( key="spend.monthly", kind=LimitKind.SPEND, floor=eur(0), grouping_defaults=partial, reserved_default=eur(0), owner="tenant-engine", ) def test_registry_floor_may_not_be_unlimited(): with pytest.raises(GuardrailRegistryInvalidError): LimitDefinition( key="entity.workspace", kind=LimitKind.ENTITY_COUNT, floor=LimitValue(kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED), grouping_defaults={ g: LimitValue(kind=LimitKind.ENTITY_COUNT, amount=1) for g in GROUPINGS }, reserved_default=LimitValue(kind=LimitKind.ENTITY_COUNT, amount=1), owner="some-service", ) def test_key_prefix_must_match_kind(): with pytest.raises(InvalidLimitError): LimitDefinition( key="entity.workspace", kind=LimitKind.SPEND, floor=eur(0), grouping_defaults={g: eur(0) for g in GROUPINGS}, reserved_default=eur(0), owner="x", ) def test_duplicate_keys_are_rejected(): with pytest.raises(GuardrailRegistryInvalidError): LimitRegistry((SPEND_MONTHLY, SPEND_MONTHLY)) def test_unregistered_key_errors_rather_than_resolving(): # not zero (would break a misspelled key) and not unlimited (fails open) with pytest.raises(UnknownLimitKeyError): resolve_limit("spend.weekly", tenant=tenant("tenant:small:acme")) with pytest.raises(UnknownLimitKeyError): DEFAULT_REGISTRY.get("entity.unregistered") def test_reserved_spend_default_is_never_unlimited(): with pytest.raises(InvalidLimitError): unlimited_reserved_default(LimitKind.SPEND) # --- Resolution: groupings ------------------------------------------------ @pytest.mark.parametrize("grouping", sorted(GROUPINGS)) def test_every_grouping_resolves_to_exactly_one_value(grouping): effective = resolve_limit("spend.monthly", tenant=tenant(f"tenant:{grouping}:x")) assert effective.provenance is Provenance.GROUPING assert isinstance(effective.value.amount, int) assert not effective.is_fail_closed def test_trial_defaults_to_zero_spend(): # ADR-0013's mandate, and the one row in the table that is canon effective = resolve_limit("spend.monthly", tenant=tenant("tenant:trial:binky")) assert effective.value.amount == 0 assert effective.provenance is Provenance.GROUPING def test_agentic_ceiling_is_tighter_than_small(): defaults = SPEND_MONTHLY.grouping_defaults assert defaults["agentic"].amount < defaults["small"].amount def test_resolve_limits_covers_the_whole_registry(): resolved = resolve_limits(tenant=tenant("tenant:medium:acme")) assert set(resolved) == set(DEFAULT_REGISTRY.keys) # --- Resolution: precedence ---------------------------------------------- def test_override_beats_plan_beats_grouping(): t = tenant("tenant:small:acme") grouping_only = resolve_limit("spend.monthly", tenant=t) assert grouping_only.provenance is Provenance.GROUPING with_plan = resolve_limit( "spend.monthly", tenant=t, plan_limits={"spend.monthly": eur(60_000)} ) assert with_plan.provenance is Provenance.PLAN assert with_plan.value.amount == 60_000 with_override = resolve_limit( "spend.monthly", tenant=t, plan_limits={"spend.monthly": eur(60_000)}, overrides={"spend.monthly": eur(1_000)}, ) assert with_override.provenance is Provenance.OVERRIDE assert with_override.value.amount == 1_000 def test_precedence_is_per_key_not_per_set(): entity = LimitDefinition( key="entity.workspace", kind=LimitKind.ENTITY_COUNT, floor=LimitValue(kind=LimitKind.ENTITY_COUNT, amount=0), grouping_defaults={g: LimitValue(kind=LimitKind.ENTITY_COUNT, amount=3) for g in GROUPINGS}, reserved_default=LimitValue(kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED), owner="some-service", ) registry = LimitRegistry((SPEND_MONTHLY, entity)) resolved = resolve_limits( tenant=tenant("tenant:small:acme"), plan_limits={"spend.monthly": eur(60_000)}, registry=registry, ) # a plan supplying spend.monthly must not wipe out the grouping-derived # entity limit assert resolved["spend.monthly"].provenance is Provenance.PLAN assert resolved["entity.workspace"].provenance is Provenance.GROUPING assert resolved["entity.workspace"].value.amount == 3 def test_unmapped_grouping_fails_closed_visibly(): # second line of defence: the registry validates totality at construction, # so this state is only reachable by bypassing it broken = object.__new__(LimitDefinition) object.__setattr__(broken, "key", "spend.monthly") object.__setattr__(broken, "kind", LimitKind.SPEND) object.__setattr__(broken, "period", "P1M") object.__setattr__(broken, "floor", eur(0)) object.__setattr__(broken, "grouping_defaults", {}) object.__setattr__(broken, "reserved_default", eur(0)) object.__setattr__(broken, "owner", "tenant-engine") effective = resolve_limit( "spend.monthly", tenant=tenant("tenant:large:acme"), registry=LimitRegistry((broken,)) ) assert effective.provenance is Provenance.FAIL_CLOSED assert effective.is_fail_closed assert effective.value.amount == 0 # --- Resolution: reserved identifiers and lifecycle ---------------------- @pytest.mark.parametrize("identifier", sorted(RESERVED_IDENTIFIERS)) def test_reserved_identifiers_use_the_reserved_profile(identifier): t = tenant(identifier) assert t.is_reserved effective = resolve_limit("spend.monthly", tenant=t) # explicit profile, not a fall-through to the floor: falling through would # clamp the platform's own identity to zero and take the platform down assert effective.provenance is Provenance.RESERVED assert effective.value.amount == 0 def test_reserved_counts_are_explicitly_unlimited(): assert unlimited_reserved_default(LimitKind.ENTITY_COUNT).is_unlimited def test_retired_tenant_clamps_to_the_floor_but_still_reads(): effective = resolve_limit("spend.monthly", tenant=tenant("tenant:large:acme", retired=True)) assert effective.provenance is Provenance.LIFECYCLE assert effective.value.amount == 0 def test_lifecycle_clamp_may_only_reduce(): # a retired tenant whose resolved value is already at the floor keeps its # provenance -- the clamp reduces, it does not relabel effective = resolve_limit("spend.monthly", tenant=tenant("tenant:trial:x", retired=True)) assert effective.value.amount == 0 assert effective.provenance is Provenance.GROUPING def test_retired_tenant_override_cannot_loosen(): effective = resolve_limit( "spend.monthly", tenant=tenant("tenant:trial:x", retired=True), overrides={"spend.monthly": eur(999_999)}, ) assert effective.value.amount == 0 assert effective.provenance is Provenance.LIFECYCLE # --- Audit --------------------------------------------------------------- def change(**kwargs) -> GuardrailChange: base = dict( change_id="c-1", tenant_id="t-1", limit_key="spend.monthly", previous=eur(0), current=eur(5_000), changed_by="ops", reason="upgrade", correlation_id="corr-1", changed_at=NOW, ) return GuardrailChange(**{**base, **kwargs}) def test_guardrail_change_carries_actor_reason_and_correlation(): c = change() assert (c.changed_by, c.reason, c.correlation_id) == ("ops", "upgrade", "corr-1") assert not c.is_clear def test_clearing_an_override_is_a_valid_change(): assert change(current=None).is_clear def test_a_change_must_actually_change_something(): with pytest.raises(InvalidLimitError): change(previous=eur(0), current=eur(0)) with pytest.raises(InvalidLimitError): change(previous=None, current=None) def test_a_change_may_not_alter_the_limit_shape(): with pytest.raises(ConflictingLimitError): change(current=LimitValue(kind=LimitKind.ENTITY_COUNT, amount=5)) def test_a_change_requires_a_reason(): with pytest.raises(InvalidLimitError): change(reason=" ")