feat(core): define organization hierarchy contracts
This commit is contained in:
@@ -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",
|
||||
]
|
||||
|
||||
136
tests/test_organization_hierarchy_contract.py
Normal file
136
tests/test_organization_hierarchy_contract.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user