user-engine/tests/test_family_dataspace_onboarding.py

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()