From c6ef6448421e381ced34dff834af0bc72998e469 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sun, 2 Aug 2026 14:44:41 +0200 Subject: [PATCH] Add typed IDM relationship contracts --- docs/MODULE_ARCHITECTURE.md | 9 +- src/govoplan_core/core/idm.py | 150 +++++++++++++++++++++++- tests/test_idm_relationship_contract.py | 79 +++++++++++++ 3 files changed, 236 insertions(+), 2 deletions(-) create mode 100644 tests/test_idm_relationship_contract.py diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index 48e7165..33ffbae 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -207,13 +207,20 @@ Other stable runtime capabilities currently include: - `identity.directory` and `identity.search` - `organizations.directory` -- `idm.directory` +- `idm.directory`, `idm.function_assignments`, `idm.relationships`, and + `idm.assignment_lifecycle` - `calendar.outbox`, `calendar.scheduling`, `calendar.invitations`, and `calendar.externalProfiles` - `poll.scheduling` - `notifications.dispatch` - `workflow.definitionContributions` and `workflow.runtimeWorker` +The provider-neutral `idm.relationships` contract carries tenant-scoped typed +groups, effective-dated identity relationships, and explicit membership +decisions. It deliberately does not expose IDM persistence models or imply an +Access permission. Consumers can retain source revisions and inclusion or +exclusion provenance while remaining optional-module safe. + Modules contribute reusable process baselines through `ModuleManifest.workflow_definitions`. Each contribution pins its origin module and version, stable key, schema and content hash, native graph/BPMN content, diff --git a/src/govoplan_core/core/idm.py b/src/govoplan_core/core/idm.py index 88961f9..ae6e80c 100644 --- a/src/govoplan_core/core/idm.py +++ b/src/govoplan_core/core/idm.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from typing import Literal, Protocol, runtime_checkable @@ -10,9 +10,12 @@ IDM_MODULE_ID = "idm" CAPABILITY_IDM_DIRECTORY = "idm.directory" CAPABILITY_IDM_FUNCTION_ASSIGNMENTS = "idm.function_assignments" CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE = "idm.assignment_lifecycle" +CAPABILITY_IDM_RELATIONSHIPS = "idm.relationships" IdmStatus = Literal["active", "inactive", "suspended"] OrganizationFunctionAssignmentSource = Literal["direct", "delegated", "acting_for", "directory", "governance", "system"] +TypedGroupStatus = Literal["active", "inactive"] +IdentityRelationshipStatus = Literal["active", "revoked"] @dataclass(frozen=True, slots=True) @@ -44,6 +47,78 @@ class OrganizationFunctionIncumbencyRef: return not self.assignments +@dataclass(frozen=True, slots=True) +class TypedGroupRef: + """Provider-neutral IDM group fact scoped to one tenant.""" + + id: str + tenant_id: str + key: str + name: str + group_type: str + description: str | None = None + status: TypedGroupStatus = "active" + source_provider: str = "local" + source_resource_type: str | None = None + source_resource_id: str | None = None + source_revision: str | None = None + properties: Mapping[str, object] = field(default_factory=dict) + provenance: Mapping[str, object] = field(default_factory=dict) + revision: int = 1 + + +@dataclass(frozen=True, slots=True) +class IdentityRelationshipRef: + """An effective-dated relationship from an identity to a typed target.""" + + id: str + tenant_id: str + relationship_kind: str + subject_identity_id: str + target_group_id: str | None = None + related_identity_id: str | None = None + role: str | None = None + valid_from: datetime | None = None + valid_until: datetime | None = None + status: IdentityRelationshipStatus = "active" + revoked_at: datetime | None = None + revoked_by: str | None = None + revocation_reason: str | None = None + source_provider: str = "local" + source_resource_type: str | None = None + source_resource_id: str | None = None + source_revision: str | None = None + properties: Mapping[str, object] = field(default_factory=dict) + provenance: Mapping[str, object] = field(default_factory=dict) + revision: int = 1 + + +@dataclass(frozen=True, slots=True) +class IdentityRelationshipDecisionRef: + relationship: IdentityRelationshipRef + included: bool + code: str + explanation: str + identity_status: IdmStatus | None = None + + +@dataclass(frozen=True, slots=True) +class TypedGroupMembershipResolutionRef: + group: TypedGroupRef + effective_at: datetime + decisions: tuple[IdentityRelationshipDecisionRef, ...] = () + + @property + def identity_ids(self) -> tuple[str, ...]: + return tuple( + dict.fromkeys( + item.relationship.subject_identity_id + for item in self.decisions + if item.included + ) + ) + + @runtime_checkable class IdmDirectory(Protocol): def get_organization_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None: @@ -109,6 +184,79 @@ class IdmFunctionAssignmentDirectory(Protocol): ... +@runtime_checkable +class IdmRelationshipDirectory(Protocol): + """Tenant-safe forward/reverse lookup for typed IDM relationships.""" + + def get_typed_group( + self, + group_id: str, + *, + tenant_id: str | None = None, + ) -> TypedGroupRef | None: + ... + + def list_typed_groups( + self, + *, + tenant_id: str, + query: str | None = None, + group_types: Sequence[str] = (), + include_inactive: bool = False, + limit: int = 100, + ) -> Sequence[TypedGroupRef]: + ... + + def identity_relationships_for_identity( + self, + identity_id: str, + *, + tenant_id: str, + effective_at: datetime | None = None, + relationship_kinds: Sequence[str] = (), + ) -> Sequence[IdentityRelationshipRef]: + ... + + def identity_relationships_for_identities( + self, + identity_ids: Sequence[str], + *, + tenant_id: str, + effective_at: datetime | None = None, + relationship_kinds: Sequence[str] = (), + ) -> Mapping[str, Sequence[IdentityRelationshipRef]]: + ... + + def identity_relationships_for_group( + self, + group_id: str, + *, + tenant_id: str, + effective_at: datetime | None = None, + relationship_kinds: Sequence[str] = (), + ) -> Sequence[IdentityRelationshipRef]: + ... + + def identity_relationships_for_groups( + self, + group_ids: Sequence[str], + *, + tenant_id: str, + effective_at: datetime | None = None, + relationship_kinds: Sequence[str] = (), + ) -> Mapping[str, Sequence[IdentityRelationshipRef]]: + ... + + def resolve_typed_group_memberships( + self, + group_ids: Sequence[str], + *, + tenant_id: str, + effective_at: datetime | None = None, + relationship_kinds: Sequence[str] = ("member",), + ) -> Mapping[str, TypedGroupMembershipResolutionRef]: + ... + @runtime_checkable class IdmAssignmentLifecycle(Protocol): """Worker boundary for time-driven function-assignment transitions.""" diff --git a/tests/test_idm_relationship_contract.py b/tests/test_idm_relationship_contract.py new file mode 100644 index 0000000..b86be87 --- /dev/null +++ b/tests/test_idm_relationship_contract.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import unittest +from datetime import datetime, timezone + +from govoplan_core.core.idm import ( + IdentityRelationshipDecisionRef, + IdentityRelationshipRef, + IdmRelationshipDirectory, + TypedGroupMembershipResolutionRef, + TypedGroupRef, +) + + +class RelationshipDirectoryStub: + def get_typed_group(self, group_id, *, tenant_id=None): + return None + + def list_typed_groups(self, **kwargs): + return () + + def identity_relationships_for_identity(self, identity_id, **kwargs): + return () + + def identity_relationships_for_identities(self, identity_ids, **kwargs): + return {identity_id: () for identity_id in identity_ids} + + def identity_relationships_for_group(self, group_id, **kwargs): + return () + + def identity_relationships_for_groups(self, group_ids, **kwargs): + return {group_id: () for group_id in group_ids} + + def resolve_typed_group_memberships(self, group_ids, **kwargs): + return {} + + +class IdmRelationshipContractTests(unittest.TestCase): + def test_runtime_protocol_and_resolution_identity_projection(self) -> None: + self.assertIsInstance(RelationshipDirectoryStub(), IdmRelationshipDirectory) + relationship = IdentityRelationshipRef( + id="relationship-1", + tenant_id="tenant-1", + relationship_kind="member", + subject_identity_id="identity-1", + target_group_id="group-1", + ) + resolution = TypedGroupMembershipResolutionRef( + group=TypedGroupRef( + id="group-1", + tenant_id="tenant-1", + key="group", + name="Group", + group_type="business_status", + ), + effective_at=datetime(2026, 8, 2, tzinfo=timezone.utc), + decisions=( + IdentityRelationshipDecisionRef( + relationship=relationship, + included=True, + code="relationship.effective", + explanation="The relationship is effective.", + identity_status="active", + ), + IdentityRelationshipDecisionRef( + relationship=relationship, + included=False, + code="relationship.revoked", + explanation="The relationship was revoked.", + identity_status="active", + ), + ), + ) + + self.assertEqual(("identity-1",), resolution.identity_ids) + + +if __name__ == "__main__": + unittest.main()