From ea436a513fad4bbe8298fa8e5ec806cffe872182 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 30 Jul 2026 03:26:19 +0200 Subject: [PATCH] feat(core): define organization hierarchy contracts --- src/govoplan_core/core/organizations.py | 318 ++++++++++++++++++ tests/test_organization_hierarchy_contract.py | 136 ++++++++ 2 files changed, 454 insertions(+) create mode 100644 tests/test_organization_hierarchy_contract.py diff --git a/src/govoplan_core/core/organizations.py b/src/govoplan_core/core/organizations.py index dcc23dc..345d221 100644 --- a/src/govoplan_core/core/organizations.py +++ b/src/govoplan_core/core/organizations.py @@ -2,13 +2,57 @@ from __future__ import annotations from collections.abc import Sequence from dataclasses import dataclass +from datetime import datetime from typing import Literal, Protocol, runtime_checkable ORGANIZATIONS_MODULE_ID = "organizations" CAPABILITY_ORGANIZATION_DIRECTORY = "organizations.directory" +CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY = ( + "organizations.hierarchyDirectory" +) OrganizationStatus = Literal["active", "inactive", "suspended"] +OrganizationResolutionStatus = Literal[ + "active", + "inactive", + "missing", + "unreachable", + "invalid", +] +OrganizationHierarchyDirection = Literal["ancestors", "descendants"] +OrganizationLifecycleResource = Literal[ + "unit_type", + "structure", + "relation_type", + "unit", + "relation", + "function_type", + "function", +] +OrganizationLifecycleAction = Literal[ + "created", + "updated", + "moved", + "deactivated", +] +ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION = 1 +ORGANIZATION_LIFECYCLE_RESOURCES: tuple[ + OrganizationLifecycleResource, + ..., +] = ( + "unit_type", + "structure", + "relation_type", + "unit", + "relation", + "function_type", + "function", +) +ORGANIZATION_LIFECYCLE_ACTIONS: tuple[ + OrganizationLifecycleAction, + ..., +] = ("created", "updated", "moved", "deactivated") @dataclass(frozen=True, slots=True) @@ -37,6 +81,134 @@ class OrganizationFunctionRef: status: OrganizationStatus = "active" +@dataclass(frozen=True, slots=True) +class OrganizationUnitTypeRef: + id: str + tenant_id: str + slug: str + name: str + description: str | None = None + status: OrganizationStatus = "active" + + +@dataclass(frozen=True, slots=True) +class OrganizationFunctionTypeRef: + id: str + tenant_id: str + slug: str + name: str + organization_unit_type_id: str | None = None + description: str | None = None + delegable: bool = False + act_in_place_allowed: bool = False + status: OrganizationStatus = "active" + + +@dataclass(frozen=True, slots=True) +class OrganizationStructureRef: + id: str + tenant_id: str + slug: str + name: str + structure_kind: str + description: str | None = None + status: OrganizationStatus = "active" + + +@dataclass(frozen=True, slots=True) +class OrganizationRelationTypeRef: + id: str + tenant_id: str + slug: str + name: str + structure_id: str | None = None + source_unit_type_id: str | None = None + target_unit_type_id: str | None = None + is_hierarchical: bool = True + allow_cycles: bool = False + description: str | None = None + status: OrganizationStatus = "active" + + +@dataclass(frozen=True, slots=True) +class OrganizationHierarchyEdgeRef: + id: str + tenant_id: str + structure: OrganizationStructureRef + relation_type: OrganizationRelationTypeRef + source_unit_id: str + target_unit_id: str + valid_from: datetime | None = None + valid_until: datetime | None = None + status: OrganizationStatus = "active" + + +@dataclass(frozen=True, slots=True) +class OrganizationHierarchyMatchRef: + unit: OrganizationUnitRef + depth: int + path: tuple[OrganizationHierarchyEdgeRef, ...] + + +@dataclass(frozen=True, slots=True) +class OrganizationHierarchyResolution: + tenant_id: str + root_unit_id: str + direction: OrganizationHierarchyDirection + structure_id: str + relation_type_ids: tuple[str, ...] + max_depth: int + status: OrganizationResolutionStatus + root: OrganizationUnitRef | None = None + matches: tuple[OrganizationHierarchyMatchRef, ...] = () + cycle_detected: bool = False + depth_limited: bool = False + diagnostics: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class OrganizationHierarchyPathResolution: + tenant_id: str + source_unit_id: str + target_unit_id: str + direction: OrganizationHierarchyDirection + structure_id: str + relation_type_ids: tuple[str, ...] + max_depth: int + status: OrganizationResolutionStatus + source: OrganizationUnitRef | None = None + target: OrganizationUnitRef | None = None + path: tuple[OrganizationHierarchyEdgeRef, ...] = () + cycle_detected: bool = False + depth_limited: bool = False + diagnostics: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class OrganizationFunctionTypeResolution: + tenant_id: str + function_type_id: str + requested_unit_ids: tuple[str, ...] + status: OrganizationResolutionStatus + function_type: OrganizationFunctionTypeRef | None = None + matches: tuple[OrganizationFunctionRef, ...] = () + missing_unit_ids: tuple[str, ...] = () + inactive_unit_ids: tuple[str, ...] = () + diagnostics: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class OrganizationUnitTypeResolution: + tenant_id: str + unit_type_id: str + status: OrganizationResolutionStatus + unit_type: OrganizationUnitTypeRef | None = None + structure_id: str | None = None + root_unit_id: str | None = None + matches: tuple[OrganizationUnitRef, ...] = () + diagnostics: tuple[str, ...] = () + + @runtime_checkable class OrganizationDirectory(Protocol): def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None: @@ -55,3 +227,149 @@ class OrganizationDirectory(Protocol): include_subunits: bool = False, ) -> Sequence[OrganizationFunctionRef]: ... + + +@runtime_checkable +class OrganizationHierarchyDirectory(Protocol): + def get_unit_type( + self, + tenant_id: str, + unit_type_id: str, + ) -> OrganizationUnitTypeRef | None: + ... + + def get_function_type( + self, + tenant_id: str, + function_type_id: str, + ) -> OrganizationFunctionTypeRef | None: + ... + + def resolve_functions_by_type( + self, + tenant_id: str, + function_type_id: str, + *, + organization_unit_ids: Sequence[str] = (), + ) -> OrganizationFunctionTypeResolution: + ... + + def resolve_units_by_type( + self, + tenant_id: str, + unit_type_id: str, + *, + structure_id: str | None = None, + root_unit_id: str | None = None, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "descendants", + max_depth: int = 10, + ) -> OrganizationUnitTypeResolution: + ... + + def resolve_hierarchy_relatives( + self, + tenant_id: str, + organization_unit_ids: Sequence[str], + *, + structure_id: str, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "ancestors", + max_depth: int = 10, + ) -> Sequence[OrganizationHierarchyResolution]: + ... + + def resolve_hierarchy_paths( + self, + tenant_id: str, + unit_pairs: Sequence[tuple[str, str]], + *, + structure_id: str, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "descendants", + max_depth: int = 10, + ) -> Sequence[OrganizationHierarchyPathResolution]: + ... + + +def organization_directory( + registry: object | None, +) -> OrganizationDirectory | None: + capability = _capability(registry, CAPABILITY_ORGANIZATION_DIRECTORY) + return capability if isinstance(capability, OrganizationDirectory) else None + + +def organization_hierarchy_directory( + registry: object | None, +) -> OrganizationHierarchyDirectory | None: + capability = _capability( + registry, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, + ) + return ( + capability + if isinstance(capability, OrganizationHierarchyDirectory) + else None + ) + + +def organization_lifecycle_event_type( + resource: OrganizationLifecycleResource, + action: OrganizationLifecycleAction, +) -> str: + if resource not in ORGANIZATION_LIFECYCLE_RESOURCES: + raise ValueError("Unsupported organization lifecycle resource.") + if action not in ORGANIZATION_LIFECYCLE_ACTIONS: + raise ValueError("Unsupported organization lifecycle action.") + return f"organizations.{resource}.{action}.v1" + + +ORGANIZATION_DIRECTORY_INVALIDATION_EVENT_TYPES = frozenset( + organization_lifecycle_event_type(resource, action) + for resource in ORGANIZATION_LIFECYCLE_RESOURCES + for action in ORGANIZATION_LIFECYCLE_ACTIONS +) + + +def _capability(registry: object | None, name: str) -> object | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(name) + ): + return None + return registry.capability(name) + + +__all__ = [ + "CAPABILITY_ORGANIZATION_DIRECTORY", + "CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY", + "ORGANIZATION_DIRECTORY_INVALIDATION_EVENT_TYPES", + "ORGANIZATION_LIFECYCLE_ACTIONS", + "ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION", + "ORGANIZATION_LIFECYCLE_RESOURCES", + "ORGANIZATIONS_MODULE_ID", + "OrganizationDirectory", + "OrganizationFunctionRef", + "OrganizationFunctionTypeRef", + "OrganizationFunctionTypeResolution", + "OrganizationHierarchyDirection", + "OrganizationHierarchyDirectory", + "OrganizationHierarchyEdgeRef", + "OrganizationHierarchyMatchRef", + "OrganizationHierarchyPathResolution", + "OrganizationHierarchyResolution", + "OrganizationLifecycleAction", + "OrganizationLifecycleResource", + "OrganizationRelationTypeRef", + "OrganizationResolutionStatus", + "OrganizationStatus", + "OrganizationStructureRef", + "OrganizationUnitRef", + "OrganizationUnitTypeRef", + "OrganizationUnitTypeResolution", + "organization_directory", + "organization_hierarchy_directory", + "organization_lifecycle_event_type", +] diff --git a/tests/test_organization_hierarchy_contract.py b/tests/test_organization_hierarchy_contract.py new file mode 100644 index 0000000..66227b0 --- /dev/null +++ b/tests/test_organization_hierarchy_contract.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.organizations import ( + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, + OrganizationDirectory, + OrganizationFunctionTypeRef, + OrganizationHierarchyDirectory, + OrganizationUnitTypeRef, + organization_hierarchy_directory, +) +from govoplan_core.core.registry import PlatformRegistry +from govoplan_core.core.modules import ModuleContext, ModuleManifest + + +class LegacyOrganizationDirectory: + def get_organization_unit(self, _organization_unit_id): + return None + + def organization_units_for_tenant(self, _tenant_id): + return () + + def get_function(self, _function_id): + return None + + def functions_for_organization_unit( + self, + _organization_unit_id, + *, + include_subunits=False, + ): + del include_subunits + return () + + +class HierarchyDirectory: + def get_unit_type(self, _tenant_id, _unit_type_id): + return None + + def get_function_type(self, _tenant_id, _function_type_id): + return None + + def resolve_functions_by_type( + self, + _tenant_id, + _function_type_id, + *, + organization_unit_ids=(), + ): + del organization_unit_ids + return None + + def resolve_units_by_type( + self, + _tenant_id, + _unit_type_id, + **_options, + ): + return None + + def resolve_hierarchy_relatives( + self, + _tenant_id, + _organization_unit_ids, + **_options, + ): + return () + + def resolve_hierarchy_paths( + self, + _tenant_id, + _unit_pairs, + **_options, + ): + return () + + +class OrganizationHierarchyContractTests(unittest.TestCase): + def test_legacy_directory_contract_is_not_broadened(self) -> None: + self.assertIsInstance( + LegacyOrganizationDirectory(), + OrganizationDirectory, + ) + self.assertNotIsInstance( + LegacyOrganizationDirectory(), + OrganizationHierarchyDirectory, + ) + + def test_hierarchy_capability_resolves_independently(self) -> None: + provider = HierarchyDirectory() + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="organization_hierarchy_test", + name="Organization hierarchy test", + version="0.0.0", + capability_factories={ + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY: ( + lambda _context: provider + ), + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + self.assertIsInstance(provider, OrganizationHierarchyDirectory) + self.assertIs( + provider, + organization_hierarchy_directory(registry), + ) + + def test_type_refs_preserve_tenant_and_status(self) -> None: + unit_type = OrganizationUnitTypeRef( + id="unit-type-1", + tenant_id="tenant-1", + slug="department", + name="Department", + status="inactive", + ) + function_type = OrganizationFunctionTypeRef( + id="function-type-1", + tenant_id="tenant-1", + slug="intake", + name="Intake", + organization_unit_type_id=unit_type.id, + ) + + self.assertEqual("inactive", unit_type.status) + self.assertEqual(unit_type.id, function_type.organization_unit_type_id) + + +if __name__ == "__main__": + unittest.main()