generated from coulomb/repo-seed
Implement family dataspace onboarding
This commit is contained in:
196
tests/test_family_dataspace_onboarding.py
Normal file
196
tests/test_family_dataspace_onboarding.py
Normal file
@@ -0,0 +1,196 @@
|
||||
import unittest
|
||||
|
||||
from user_engine.adapters.local import InMemoryUserEngineStore
|
||||
from user_engine.domain import (
|
||||
AccountStatus,
|
||||
FamilyDataspaceRequest,
|
||||
FamilyMemberSpec,
|
||||
FamilyRole,
|
||||
InvitationStatus,
|
||||
)
|
||||
from user_engine.errors import AuthorizationDenied, ValidationError
|
||||
from user_engine.service import UserEngineService
|
||||
from user_engine.testing.fixtures import human_actor_claims
|
||||
from user_engine.testing.scenarios import (
|
||||
ScenarioAuthorizationHarness,
|
||||
StrictFixtureIdentityClaimsAdapter,
|
||||
)
|
||||
|
||||
|
||||
class FamilyDataspaceOnboardingTests(unittest.TestCase):
|
||||
def test_onboarding_creates_family_scope_dataspace_app_and_invitation(self):
|
||||
service, store, authz = _service()
|
||||
session = service.me(_owner_claims(), correlation_id="corr-owner")
|
||||
|
||||
onboarding = service.onboard_family_dataspace(
|
||||
session.actor,
|
||||
FamilyDataspaceRequest(
|
||||
tenant="tenant:worsch-family",
|
||||
family_scope_id="family:worsch",
|
||||
family_display_name="Worsch Family",
|
||||
application_id="app.personal-dataspace",
|
||||
oidc_client_id="personal-dataspace-client",
|
||||
protected_system_id="dataspace.personal.worsch",
|
||||
member_specs=(
|
||||
FamilyMemberSpec(
|
||||
primary_email="child@example.test",
|
||||
display_name="Child Member",
|
||||
role=FamilyRole.CHILD,
|
||||
),
|
||||
),
|
||||
),
|
||||
correlation_id="corr-family-onboard",
|
||||
)
|
||||
|
||||
self.assertEqual(onboarding.tenant, "tenant:worsch-family")
|
||||
self.assertEqual(onboarding.binding.oidc_client_id, "personal-dataspace-client")
|
||||
self.assertEqual(onboarding.catalog.namespace, "dataspace")
|
||||
self.assertEqual(onboarding.owner_membership.kind, FamilyRole.OWNER.value)
|
||||
self.assertEqual(onboarding.invitations[0].invitation.status, InvitationStatus.PENDING)
|
||||
self.assertEqual(onboarding.invitations[0].tenant_account.status, AccountStatus.INVITED)
|
||||
self.assertEqual(onboarding.identity_context.entity_refs["family:family:worsch"].concept, "Group")
|
||||
self.assertEqual(
|
||||
onboarding.claims_projection.values["dataspace.family_display_name"],
|
||||
"Worsch Family",
|
||||
)
|
||||
self.assertIn("family_dataspace.onboarded", _event_types(service))
|
||||
self.assertIn("family_member.invited", _event_types(service))
|
||||
self.assertIn("family_dataspace.onboard", [request.action for request in authz.requests])
|
||||
self.assertEqual(len(store.family_invitations), 1)
|
||||
|
||||
def test_member_acceptance_links_sso_identity_and_returns_dataspace_context(self):
|
||||
service, _, _ = _service()
|
||||
owner = service.me(_owner_claims(), correlation_id="corr-owner")
|
||||
onboarding = _onboard_family(service, owner.actor)
|
||||
invitation = onboarding.invitations[0].invitation
|
||||
|
||||
acceptance = service.accept_family_invitation(
|
||||
_member_claims(subject="child-sso"),
|
||||
invitation.invitation_id,
|
||||
correlation_id="corr-accept",
|
||||
)
|
||||
|
||||
self.assertEqual(acceptance.invitation.status, InvitationStatus.ACCEPTED)
|
||||
self.assertEqual(acceptance.session.user.user_id, invitation.user_id)
|
||||
self.assertEqual(acceptance.session.account.status, AccountStatus.ACTIVE)
|
||||
self.assertEqual(
|
||||
service.store.tenant_account("tenant:worsch-family", invitation.user_id).status,
|
||||
AccountStatus.ACTIVE,
|
||||
)
|
||||
self.assertEqual(
|
||||
service.store.find_identity(
|
||||
"https://issuer.example.test",
|
||||
"child-sso",
|
||||
).user_id,
|
||||
invitation.user_id,
|
||||
)
|
||||
self.assertEqual(
|
||||
acceptance.claims_projection.values["dataspace.member_display_name"],
|
||||
"Child Member",
|
||||
)
|
||||
self.assertEqual(acceptance.identity_context.memberships[0].kind, FamilyRole.CHILD.value)
|
||||
self.assertIn("family_invitation.accepted", _event_types(service))
|
||||
|
||||
def test_revoked_invitation_cannot_be_accepted(self):
|
||||
service, _, _ = _service()
|
||||
owner = service.me(_owner_claims(), correlation_id="corr-owner")
|
||||
onboarding = _onboard_family(service, owner.actor)
|
||||
invitation = onboarding.invitations[0].invitation
|
||||
|
||||
resent = service.resend_family_invitation(
|
||||
owner.actor,
|
||||
invitation.invitation_id,
|
||||
correlation_id="corr-resend",
|
||||
)
|
||||
revoked = service.revoke_family_invitation(
|
||||
owner.actor,
|
||||
invitation.invitation_id,
|
||||
correlation_id="corr-revoke",
|
||||
)
|
||||
|
||||
self.assertEqual(resent.resend_count, 1)
|
||||
self.assertEqual(revoked.status, InvitationStatus.REVOKED)
|
||||
self.assertEqual(
|
||||
service.store.tenant_account("tenant:worsch-family", invitation.user_id).status,
|
||||
AccountStatus.DISABLED,
|
||||
)
|
||||
with self.assertRaises(ValidationError):
|
||||
service.accept_family_invitation(
|
||||
_member_claims(subject="revoked-child"),
|
||||
invitation.invitation_id,
|
||||
correlation_id="corr-revoked-accept",
|
||||
)
|
||||
|
||||
def test_cross_tenant_invitation_acceptance_is_denied(self):
|
||||
service, _, _ = _service()
|
||||
owner = service.me(_owner_claims(), correlation_id="corr-owner")
|
||||
onboarding = _onboard_family(service, owner.actor)
|
||||
|
||||
with self.assertRaises(AuthorizationDenied):
|
||||
service.accept_family_invitation(
|
||||
_member_claims(subject="wrong-tenant", tenant="tenant:other-family"),
|
||||
onboarding.invitations[0].invitation.invitation_id,
|
||||
correlation_id="corr-wrong-tenant",
|
||||
)
|
||||
|
||||
|
||||
def _service():
|
||||
store = InMemoryUserEngineStore()
|
||||
service = UserEngineService(
|
||||
store=store,
|
||||
identity_adapter=StrictFixtureIdentityClaimsAdapter(),
|
||||
authorization=ScenarioAuthorizationHarness(),
|
||||
)
|
||||
return service, store, service.authorization
|
||||
|
||||
|
||||
def _onboard_family(service: UserEngineService, actor):
|
||||
return service.onboard_family_dataspace(
|
||||
actor,
|
||||
FamilyDataspaceRequest(
|
||||
tenant="tenant:worsch-family",
|
||||
family_scope_id="family:worsch",
|
||||
family_display_name="Worsch Family",
|
||||
application_id="app.personal-dataspace",
|
||||
oidc_client_id="personal-dataspace-client",
|
||||
protected_system_id="dataspace.personal.worsch",
|
||||
member_specs=(
|
||||
FamilyMemberSpec(
|
||||
primary_email="child@example.test",
|
||||
display_name="Child Member",
|
||||
role=FamilyRole.CHILD,
|
||||
),
|
||||
),
|
||||
),
|
||||
correlation_id="corr-family-onboard",
|
||||
)
|
||||
|
||||
|
||||
def _owner_claims() -> dict[str, object]:
|
||||
claims = human_actor_claims(
|
||||
subject="family-owner",
|
||||
tenant="tenant:worsch-family",
|
||||
)
|
||||
claims["roles"] = ["tenant-admin"]
|
||||
claims["preferred_username"] = "family.owner"
|
||||
claims["email"] = "owner@example.test"
|
||||
return claims
|
||||
|
||||
|
||||
def _member_claims(
|
||||
*,
|
||||
subject: str,
|
||||
tenant: str = "tenant:worsch-family",
|
||||
) -> dict[str, object]:
|
||||
claims = human_actor_claims(subject=subject, tenant=tenant)
|
||||
claims["preferred_username"] = subject
|
||||
claims["email"] = f"{subject}@example.test"
|
||||
return claims
|
||||
|
||||
|
||||
def _event_types(service: UserEngineService) -> list[str]:
|
||||
return [event.event_type for event in service.outbox_events()]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -44,6 +44,7 @@ class IntegratedScenarioTests(unittest.TestCase):
|
||||
"two_applications",
|
||||
"sensitive_redaction",
|
||||
"audit_event_replay",
|
||||
"family_dataspace_onboarding",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user