generated from coulomb/repo-seed
Add tenant-aware user-engine behavior
This commit is contained in:
218
tests/test_multi_tenancy.py
Normal file
218
tests/test_multi_tenancy.py
Normal file
@@ -0,0 +1,218 @@
|
||||
import unittest
|
||||
|
||||
from user_engine.adapters.local import (
|
||||
InMemoryUserEngineStore,
|
||||
LocalAuthorizationCheckPort,
|
||||
)
|
||||
from user_engine.domain import AccountStatus, ProfileScope
|
||||
from user_engine.errors import AuthorizationDenied
|
||||
from user_engine.service import PLATFORM_TENANT, UserEngineService
|
||||
from user_engine.testing.fixtures import (
|
||||
FixtureIdentityClaimsAdapter,
|
||||
human_actor_claims,
|
||||
sample_application,
|
||||
sample_application_binding,
|
||||
sample_catalog,
|
||||
)
|
||||
|
||||
|
||||
class MultiTenancyTests(unittest.TestCase):
|
||||
def test_tenant_context_denies_cross_tenant_and_platform_root(self):
|
||||
service, _, _ = _service()
|
||||
session = service.me(_claims(), correlation_id="corr-me")
|
||||
|
||||
with self.assertRaises(AuthorizationDenied):
|
||||
service.resolve_tenant_context(session.actor, "tenant:faraday")
|
||||
|
||||
with self.assertRaises(AuthorizationDenied):
|
||||
service.resolve_tenant_context(session.actor, PLATFORM_TENANT)
|
||||
|
||||
def test_tenant_account_membership_and_diagnostics(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap(service)
|
||||
|
||||
account = service.set_tenant_account_status(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
AccountStatus.ACTIVE,
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-tenant-account",
|
||||
)
|
||||
membership = service.add_membership(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
scope_type="team",
|
||||
scope_id="team:demo",
|
||||
kind="admin",
|
||||
correlation_id="corr-membership",
|
||||
)
|
||||
diagnostics = service.tenant_diagnostics(
|
||||
session.actor,
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-diagnostics",
|
||||
)
|
||||
|
||||
self.assertEqual(account.tenant, "tenant:coulomb")
|
||||
self.assertEqual(membership.tenant, "tenant:coulomb")
|
||||
self.assertTrue(diagnostics.checks["membership_facts"])
|
||||
self.assertNotIn("membership_facts", diagnostics.issues)
|
||||
self.assertIn(
|
||||
"membership.added",
|
||||
[event.event_type for event in service.outbox_events()],
|
||||
)
|
||||
|
||||
def test_tenant_profile_precedes_global_and_app_precedes_tenant(self):
|
||||
service, _, _ = _service()
|
||||
session = _bootstrap(service)
|
||||
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"compact",
|
||||
correlation_id="corr-global",
|
||||
)
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"comfortable",
|
||||
scope=ProfileScope.TENANT,
|
||||
scope_id="tenant:coulomb",
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-tenant-profile",
|
||||
)
|
||||
tenant_profile = service.effective_profile(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-effective-tenant",
|
||||
)
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"compact",
|
||||
scope=ProfileScope.APPLICATION,
|
||||
scope_id="app.demo",
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-app-profile",
|
||||
)
|
||||
app_profile = service.effective_profile(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
application_id="app.demo",
|
||||
correlation_id="corr-effective-app",
|
||||
)
|
||||
|
||||
self.assertEqual(tenant_profile.values["demo.display_density"], "comfortable")
|
||||
self.assertEqual(
|
||||
tenant_profile.source_layers["demo.display_density"],
|
||||
"tenant:tenant:coulomb",
|
||||
)
|
||||
self.assertEqual(app_profile.values["demo.display_density"], "compact")
|
||||
self.assertEqual(
|
||||
app_profile.source_layers["demo.display_density"],
|
||||
"application:app.demo",
|
||||
)
|
||||
|
||||
def test_authorization_context_includes_membership_and_assurance_facts(self):
|
||||
service, _, authz = _service()
|
||||
session = _bootstrap(service)
|
||||
service.add_membership(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
tenant="tenant:coulomb",
|
||||
scope_type="team",
|
||||
scope_id="team:demo",
|
||||
kind="member",
|
||||
correlation_id="corr-membership",
|
||||
)
|
||||
|
||||
service.set_profile_value(
|
||||
session.actor,
|
||||
session.user.user_id,
|
||||
"demo.display_density",
|
||||
"compact",
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-profile",
|
||||
)
|
||||
|
||||
request = authz.requests[-1]
|
||||
self.assertEqual(request.tenant, "tenant:coulomb")
|
||||
self.assertIn("tenant-admin", request.context["actor_roles"])
|
||||
self.assertTrue(request.context["assurance"]["mfa"])
|
||||
self.assertEqual(len(request.context["target_memberships"]), 1)
|
||||
|
||||
def test_platform_operator_can_manage_another_tenant(self):
|
||||
service, _, _ = _service()
|
||||
platform_session = service.me(
|
||||
_claims(
|
||||
subject="platform-operator",
|
||||
tenant=PLATFORM_TENANT,
|
||||
roles=("platform-operator",),
|
||||
),
|
||||
correlation_id="corr-platform-me",
|
||||
)
|
||||
managed_user = service.create_user(
|
||||
platform_session.actor,
|
||||
display_name="Managed User",
|
||||
primary_email="managed@example.test",
|
||||
correlation_id="corr-create-managed",
|
||||
)
|
||||
|
||||
account = service.set_tenant_account_status(
|
||||
platform_session.actor,
|
||||
managed_user.user_id,
|
||||
AccountStatus.ACTIVE,
|
||||
tenant="tenant:coulomb",
|
||||
correlation_id="corr-platform-tenant-account",
|
||||
)
|
||||
|
||||
self.assertEqual(account.tenant, "tenant:coulomb")
|
||||
self.assertEqual(account.status, AccountStatus.ACTIVE)
|
||||
|
||||
|
||||
def _service():
|
||||
store = InMemoryUserEngineStore()
|
||||
authz = LocalAuthorizationCheckPort()
|
||||
service = UserEngineService(
|
||||
store=store,
|
||||
identity_adapter=FixtureIdentityClaimsAdapter(),
|
||||
authorization=authz,
|
||||
)
|
||||
return service, store, authz
|
||||
|
||||
|
||||
def _bootstrap(service: UserEngineService):
|
||||
session = service.me(_claims(), correlation_id="corr-me")
|
||||
service.register_application(
|
||||
session.actor,
|
||||
sample_application(),
|
||||
binding=sample_application_binding(),
|
||||
correlation_id="corr-app",
|
||||
)
|
||||
service.publish_catalog(
|
||||
session.actor,
|
||||
sample_catalog(),
|
||||
correlation_id="corr-catalog",
|
||||
)
|
||||
return session
|
||||
|
||||
|
||||
def _claims(
|
||||
*,
|
||||
subject: str = "tenant-admin",
|
||||
tenant: str = "tenant:coulomb",
|
||||
roles: tuple[str, ...] = ("tenant-admin",),
|
||||
):
|
||||
claims = human_actor_claims(subject=subject, tenant=tenant)
|
||||
claims["roles"] = list(roles)
|
||||
return claims
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user