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
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""TEN-WP-0011-T05: events_for is tenant-scoped; no unfiltered dump."""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from tenant_engine.domain import CapabilityRole, PlanAssignment, Tenant, create_role_grant
|
|
from tenant_engine.store import InMemoryTenantStore
|
|
|
|
|
|
def _tenant(store: InMemoryTenantStore, tenant_id: str, identifier: str) -> Tenant:
|
|
tenant = Tenant.create(tenant_id=tenant_id, identifier=identifier)
|
|
store.create_tenant(tenant)
|
|
return tenant
|
|
|
|
|
|
def test_events_for_does_not_return_another_tenants_events():
|
|
store = InMemoryTenantStore()
|
|
a = _tenant(store, "t-a", "tenant:friendly:alpha")
|
|
b = _tenant(store, "t-b", "tenant:friendly:beta")
|
|
store.grant_role(
|
|
create_role_grant(
|
|
tenant=a,
|
|
grant_id="g-a",
|
|
role=CapabilityRole.CUS,
|
|
grant_reason="manual_grant",
|
|
plan_id=None,
|
|
granted_by="ops",
|
|
correlation_id="c-a",
|
|
granted_at=datetime.now(UTC),
|
|
)
|
|
)
|
|
store.assign_plan(
|
|
PlanAssignment(tenant_id=b.tenant_id, plan_id="plan-b", assigned_at=datetime.now(UTC))
|
|
)
|
|
|
|
a_events = store.events_for(a.tenant_id)
|
|
b_events = store.events_for(b.tenant_id)
|
|
assert all(event.tenant_id == a.tenant_id for event in a_events)
|
|
assert all(event.tenant_id == b.tenant_id for event in b_events)
|
|
assert any(e.event_type == "role_granted" for e in a_events)
|
|
assert not any(e.event_type == "role_granted" for e in b_events)
|
|
assert any(e.event_type == "plan_assigned" for e in b_events)
|
|
assert not any(e.event_type == "plan_assigned" for e in a_events)
|
|
|
|
|
|
def test_production_protocol_has_no_unfiltered_events():
|
|
assert not hasattr(InMemoryTenantStore, "events") or not callable(
|
|
getattr(InMemoryTenantStore(), "events", None)
|
|
)
|
|
store = InMemoryTenantStore()
|
|
with pytest.raises(AttributeError):
|
|
store.events() # type: ignore[attr-defined]
|
|
|
|
|
|
def test_events_for_unknown_tenant_is_not_found():
|
|
store = InMemoryTenantStore()
|
|
from tenant_engine.store import TenantNotFoundError
|
|
|
|
with pytest.raises(TenantNotFoundError):
|
|
store.events_for("does-not-exist")
|