GovOPlaN/govoplan-postbox
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: route bounded hierarchy postbox copies

Browse Source
This commit is contained in:
zemion
2026-07-30 04:00:31 +02:00
parent d848a9a503
commit 28b60782de
16 changed files with 2715 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_manifest.py
+2
View File
@@ -8,6 +8,7 @@ from govoplan_core.core.postbox import (
CAPABILITY_POSTBOX_DIRECTORY,
CAPABILITY_POSTBOX_EVIDENCE,
CAPABILITY_POSTBOX_MESSAGES,
CAPABILITY_POSTBOX_ROUTING,
)
from govoplan_postbox.backend.manifest import get_manifest
@@ -28,6 +29,7 @@ class PostboxManifestTests(unittest.TestCase):
CAPABILITY_POSTBOX_MESSAGES,
CAPABILITY_POSTBOX_DELIVERY,
CAPABILITY_POSTBOX_EVIDENCE,
CAPABILITY_POSTBOX_ROUTING,
},
set(manifest.capability_factories),
)
tests/test_migration.py
+20
View File
@@ -14,14 +14,21 @@ class PostboxMigrationTests(unittest.TestCase):
"govoplan_postbox.backend.migrations.versions."
"c7d2e5f8a1b4_v010_postbox_baseline"
)
route_migration = importlib.import_module(
"govoplan_postbox.backend.migrations.versions."
"e4b7c9d2a6f1_v011_hierarchy_routes"
)
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
operations = Operations(MigrationContext.configure(connection))
original = migration.op
route_original = route_migration.op
migration.op = operations
route_migration.op = operations
try:
migration.upgrade()
route_migration.upgrade()
tables = set(inspect(connection).get_table_names())
self.assertIn("postboxes", tables)
self.assertIn("postbox_messages", tables)
@@ -43,6 +50,18 @@ class PostboxMigrationTests(unittest.TestCase):
"withdrawn_at",
}.issubset(message_columns)
)
route_columns = {
column["name"]
for column in inspect(connection).get_columns(
"postbox_routes"
)
}
self.assertTrue(
{"execute_after", "processed_at"}.issubset(
route_columns
)
)
route_migration.downgrade()
migration.downgrade()
self.assertFalse(
{
@@ -53,6 +72,7 @@ class PostboxMigrationTests(unittest.TestCase):
)
finally:
migration.op = original
route_migration.op = route_original
finally:
engine.dispose()
tests/test_router.py
+17
View File
@@ -297,6 +297,23 @@ class PostboxRouterTests(unittest.TestCase):
self.assertEqual(200, unread.status_code, unread.text)
self.assertEqual(0, unread.json()["total"])
def test_routing_dry_run_explains_default_disabled_state(self) -> None:
response = self.client.post(
"/api/v1/postbox/routing/dry-run",
json={
"target": {"postbox_id": self.postbox_id},
"producer_module": "campaigns",
"classification": "internal",
},
)
self.assertEqual(200, response.status_code, response.text)
self.assertEqual("disabled", response.json()["status"])
self.assertEqual(
["hierarchy_routing_disabled"],
response.json()["diagnostics"],
)
if __name__ == "__main__":
unittest.main()
tests/test_service.py
+630
View File
@@ -14,7 +14,17 @@ from govoplan_core.core.idm import (
)
from govoplan_core.core.organizations import (
OrganizationFunctionRef,
OrganizationFunctionTypeRef,
OrganizationFunctionTypeResolution,
OrganizationHierarchyCatalogRef,
OrganizationHierarchyEdgeRef,
OrganizationHierarchyMatchRef,
OrganizationHierarchyPathResolution,
OrganizationHierarchyResolution,
OrganizationRelationTypeRef,
OrganizationStructureRef,
OrganizationUnitRef,
OrganizationUnitTypeResolution,
)
from govoplan_core.core.postbox import (
PostboxActorRef,
@@ -173,6 +183,13 @@ class FakeOrganizationDirectory:
unit_type_id="desk",
parent_id="unit-1",
),
"unit-top": OrganizationUnitRef(
id="unit-top",
tenant_id="tenant-1",
slug="central-office",
name="Central Office",
unit_type_id="office",
),
}
self.functions = {
"function-1": OrganizationFunctionRef(
@@ -193,7 +210,47 @@ class FakeOrganizationDirectory:
function_type_id="case-clerk-type",
delegable=True,
),
"function-top": OrganizationFunctionRef(
id="function-top",
tenant_id="tenant-1",
organization_unit_id="unit-top",
slug="case-clerk",
name="Case Clerk",
function_type_id="case-clerk-type",
delegable=True,
),
}
self.structure = OrganizationStructureRef(
id="structure-1",
tenant_id="tenant-1",
slug="administrative",
name="Administrative hierarchy",
structure_kind="administrative",
)
self.parallel_structure = OrganizationStructureRef(
id="structure-2",
tenant_id="tenant-1",
slug="reporting",
name="Reporting hierarchy",
structure_kind="reporting",
)
self.relation_type = OrganizationRelationTypeRef(
id="relation-type-1",
tenant_id="tenant-1",
slug="reports-to",
name="Reports to",
structure_id=self.structure.id,
)
self.parallel_relation_type = OrganizationRelationTypeRef(
id="relation-type-2",
tenant_id="tenant-1",
slug="reports-to",
name="Reports to",
structure_id=self.parallel_structure.id,
)
self.last_hierarchy_request: dict[str, object] = {}
self.duplicate_parent_match = False
self.cycle_detected = False
def get_organization_unit(self, organization_unit_id: str):
return self.units.get(organization_unit_id)
@@ -218,6 +275,214 @@ class FakeOrganizationDirectory:
if function.organization_unit_id == organization_unit_id
)
def hierarchy_catalog(self, tenant_id: str):
if tenant_id != "tenant-1":
return OrganizationHierarchyCatalogRef(tenant_id=tenant_id)
return OrganizationHierarchyCatalogRef(
tenant_id=tenant_id,
structures=(self.structure, self.parallel_structure),
relation_types=(
self.relation_type,
self.parallel_relation_type,
),
)
def get_unit_type(self, tenant_id: str, unit_type_id: str):
del tenant_id, unit_type_id
return None
def get_function_type(self, tenant_id: str, function_type_id: str):
if tenant_id != "tenant-1" or function_type_id != "case-clerk-type":
return None
return OrganizationFunctionTypeRef(
id=function_type_id,
tenant_id=tenant_id,
slug="case-clerk",
name="Case Clerk",
)
def resolve_functions_by_type(
self,
tenant_id: str,
function_type_id: str,
*,
organization_unit_ids=(),
):
if tenant_id != "tenant-1":
return OrganizationFunctionTypeResolution(
tenant_id=tenant_id,
function_type_id=function_type_id,
requested_unit_ids=tuple(organization_unit_ids),
status="missing",
)
return OrganizationFunctionTypeResolution(
tenant_id=tenant_id,
function_type_id=function_type_id,
requested_unit_ids=tuple(organization_unit_ids),
status="active",
function_type=self.get_function_type(
tenant_id,
function_type_id,
),
matches=tuple(
function
for function in self.functions.values()
if function.function_type_id == function_type_id
and (
not organization_unit_ids
or function.organization_unit_id
in organization_unit_ids
)
),
)
def resolve_units_by_type(
self,
tenant_id: str,
unit_type_id: str,
**kwargs,
):
del kwargs
return OrganizationUnitTypeResolution(
tenant_id=tenant_id,
unit_type_id=unit_type_id,
status="active",
matches=tuple(
unit
for unit in self.units.values()
if unit.tenant_id == tenant_id
and unit.unit_type_id == unit_type_id
),
)
def resolve_hierarchy_relatives(
self,
tenant_id: str,
organization_unit_ids,
*,
structure_id: str,
relation_type_ids=(),
direction="ancestors",
max_depth=10,
):
self.last_hierarchy_request = {
"tenant_id": tenant_id,
"structure_id": structure_id,
"relation_type_ids": tuple(relation_type_ids),
"direction": direction,
"max_depth": max_depth,
}
results = []
for root_id in organization_unit_ids:
root = self.units.get(root_id)
if root is None or root.tenant_id != tenant_id:
results.append(
OrganizationHierarchyResolution(
tenant_id=tenant_id,
root_unit_id=root_id,
direction=direction,
structure_id=structure_id,
relation_type_ids=tuple(relation_type_ids),
max_depth=max_depth,
status="missing",
)
)
continue
if structure_id == self.parallel_structure.id:
matches = ()
else:
first_edge = OrganizationHierarchyEdgeRef(
id="edge-child-parent",
tenant_id=tenant_id,
structure=self.structure,
relation_type=self.relation_type,
source_unit_id=root_id,
target_unit_id="unit-1",
)
second_edge = OrganizationHierarchyEdgeRef(
id="edge-parent-top",
tenant_id=tenant_id,
structure=self.structure,
relation_type=self.relation_type,
source_unit_id="unit-1",
target_unit_id="unit-top",
)
available = (
OrganizationHierarchyMatchRef(
unit=self.units["unit-1"],
depth=1,
path=(first_edge,),
),
OrganizationHierarchyMatchRef(
unit=self.units["unit-top"],
depth=2,
path=(first_edge, second_edge),
),
)
if self.duplicate_parent_match:
duplicate_edge = OrganizationHierarchyEdgeRef(
id="edge-child-parent-duplicate",
tenant_id=tenant_id,
structure=self.structure,
relation_type=self.relation_type,
source_unit_id=root_id,
target_unit_id="unit-1",
)
available = (
available[0],
OrganizationHierarchyMatchRef(
unit=self.units["unit-1"],
depth=1,
path=(duplicate_edge,),
),
available[1],
)
matches = tuple(
match for match in available if match.depth <= max_depth
)
results.append(
OrganizationHierarchyResolution(
tenant_id=tenant_id,
root_unit_id=root_id,
direction=direction,
structure_id=structure_id,
relation_type_ids=tuple(relation_type_ids),
max_depth=max_depth,
status="active",
root=root,
matches=matches,
cycle_detected=self.cycle_detected,
depth_limited=(
structure_id == self.structure.id and max_depth < 2
),
)
)
return tuple(results)
def resolve_hierarchy_paths(
self,
tenant_id: str,
unit_pairs,
*,
structure_id: str,
relation_type_ids=(),
direction="descendants",
max_depth=10,
):
return tuple(
OrganizationHierarchyPathResolution(
tenant_id=tenant_id,
source_unit_id=source_id,
target_unit_id=target_id,
direction=direction,
structure_id=structure_id,
relation_type_ids=tuple(relation_type_ids),
max_depth=max_depth,
status="unreachable",
)
for source_id, target_id in unit_pairs
)
class FakeNotificationDispatch:
def __init__(self) -> None:
@@ -280,6 +545,117 @@ class PostboxServiceTests(unittest.TestCase):
actor_id="admin-1",
)
def _routing_policy(
self,
target_template_id: str,
*,
structure_id: str = "structure-1",
max_depth: int = 2,
producer_modules: tuple[str, ...] = ("campaigns",),
classifications: tuple[str, ...] = ("internal",),
vacancy_escalation: bool = False,
) -> dict[str, object]:
return {
"linked_copy": {
"enabled": True,
"structure_id": structure_id,
"relation_type_ids": ["relation-type-1"],
"max_depth": max_depth,
"target_function_type_id": "case-clerk-type",
"target_template_id": target_template_id,
"fanout": "nearest",
"allowed_classifications": list(classifications),
"allowed_producer_modules": list(producer_modules),
"require_expiry": True,
"max_retention_days": 30,
},
"attention": {
"mode": (
"vacancy_escalation"
if vacancy_escalation
else "none"
),
"delay_minutes": 1 if vacancy_escalation else None,
},
"shared_visibility": {"mode": "none"},
}
def _create_routing_source(
self,
session: Session,
*,
vacancy_escalation: bool = False,
structure_id: str = "structure-1",
max_depth: int = 2,
) -> tuple[PostboxService, Postbox, PostboxTemplate]:
service = PostboxService(
identities=FakeIdentityDirectory(), # type: ignore[arg-type]
idm=self.idm, # type: ignore[arg-type]
incumbencies=self.idm, # type: ignore[arg-type]
organizations=self.organizations, # type: ignore[arg-type]
hierarchy=self.organizations, # type: ignore[arg-type]
)
target_template = service.create_template(
session,
tenant_id="tenant-1",
slug="hierarchy-target",
name="Hierarchy target",
description=None,
function_type_id="case-clerk-type",
scope_kind="tenant",
scope_id=None,
name_pattern="{unit_name} / {function_name} Routed",
address_pattern="routed.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
actor_id="admin-1",
)
service.publish_template(
session,
tenant_id="tenant-1",
template_id=target_template.id,
revision_number=None,
actor_id="admin-1",
)
source_template = service.create_template(
session,
tenant_id="tenant-1",
slug="hierarchy-source",
name="Hierarchy source",
description=None,
function_type_id="case-clerk-type",
scope_kind="tenant",
scope_id=None,
name_pattern="{unit_name} / {function_name} Source",
address_pattern="source.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
actor_id="admin-1",
routing_policy=self._routing_policy(
target_template.id,
vacancy_escalation=vacancy_escalation,
structure_id=structure_id,
max_depth=max_depth,
),
)
service.publish_template(
session,
tenant_id="tenant-1",
template_id=source_template.id,
revision_number=None,
actor_id="admin-1",
)
source = service.materialize_template(
session,
tenant_id="tenant-1",
template_id=source_template.id,
organization_unit_id="unit-child",
function_id="function-child",
context_key="case-42",
actor_id="admin-1",
)
return service, source, target_template
def test_access_follows_current_assignment_and_reports_vacancy(self) -> None:
with Session(self.engine) as session:
postbox = self._create_exact(session)
@@ -595,6 +971,260 @@ class PostboxServiceTests(unittest.TestCase):
[event.type for event in events],
)
def test_hierarchy_linked_copy_snapshots_path_and_independent_state(
self,
) -> None:
self.idm.assignments.append(self.assignment)
with Session(self.engine) as session:
service, source, _target_template = (
self._create_routing_source(session)
)
result = service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-1",
idempotency_key="routing-copy-1",
subject="Routed decision",
body_text="The decision is available.",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
),
)
session.commit()
routes = session.query(PostboxRoute).all()
self.assertEqual(1, len(routes))
route = routes[0]
self.assertEqual("linked_copy", route.route_kind)
self.assertEqual("accepted", route.status)
self.assertNotEqual(result.message_id, route.target_message_id)
self.assertEqual(
["edge-child-parent"],
[
edge["edge_id"]
for edge in route.policy_snapshot["target"]["path"]
],
)
self.assertEqual(
route.id,
result.evidence["hierarchy_routes"][0]["route_id"],
)
service.mark_message(
session,
tenant_id="tenant-1",
message_id=route.target_message_id,
actor=self.actor,
state="read",
)
session.commit()
receipts = session.query(PostboxMessageReceipt).all()
self.assertEqual([route.target_message_id], [item.message_id for item in receipts])
def test_hierarchy_dry_run_explains_gates_depth_and_parallel_structure(
self,
) -> None:
with Session(self.engine) as session:
service, source, _target_template = (
self._create_routing_source(session, max_depth=1)
)
blocked = service.preview_hierarchy_routes(
session,
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="mail",
classification="restricted",
expires_at=None,
)
self.assertEqual("blocked", blocked["status"])
self.assertTrue(
{
"classification_not_allowed",
"producer_not_authorized",
"expiry_required",
}.issubset(set(blocked["diagnostics"]))
)
planned = service.preview_hierarchy_routes(
session,
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
)
self.assertEqual("planned", planned["status"])
self.assertEqual(1, len(planned["routes"]))
self.assertEqual(
1,
self.organizations.last_hierarchy_request["max_depth"],
)
self.assertIn(
"hierarchy_depth_limited",
planned["diagnostics"],
)
with Session(self.engine) as session:
service, source, _target_template = self._create_routing_source(
session,
structure_id="structure-2",
)
no_route = service.preview_hierarchy_routes(
session,
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
)
self.assertEqual("no_route", no_route["status"])
self.assertIn("no_hierarchy_ancestor", no_route["diagnostics"])
def test_hierarchy_preview_bounds_cycles_deduplicates_and_is_tenant_safe(
self,
) -> None:
self.organizations.duplicate_parent_match = True
self.organizations.cycle_detected = True
with Session(self.engine) as session:
service, source, _target_template = (
self._create_routing_source(session)
)
preview = service.preview_hierarchy_routes(
session,
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
)
self.assertIn("hierarchy_cycle_bounded", preview["diagnostics"])
self.assertEqual(
1,
sum(
route["status"] == "duplicate"
for route in preview["routes"]
),
)
with self.assertRaisesRegex(
ValueError,
"existing source Postbox",
):
service.preview_hierarchy_routes(
session,
tenant_id="tenant-2",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
)
def test_vacancy_escalation_is_delayed_and_uses_frozen_targets(
self,
) -> None:
top_assignment = OrganizationFunctionAssignmentRef(
id="assignment-top",
tenant_id="tenant-1",
identity_id="identity-top",
account_id="account-top",
function_id="function-top",
organization_unit_id="unit-top",
source="direct",
)
self.idm.assignments.append(top_assignment)
with Session(self.engine) as session:
service, source, _target_template = self._create_routing_source(
session,
vacancy_escalation=True,
)
service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-2",
idempotency_key="routing-vacancy-1",
subject="Escalated decision",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
),
)
session.commit()
routes = (
session.query(PostboxRoute)
.order_by(PostboxRoute.depth)
.all()
)
self.assertEqual(
["accepted_vacant", "pending_vacancy_escalation"],
[route.status for route in routes],
)
pending = routes[1]
frozen_target_id = pending.target_postbox_id
pending.execute_after = utc_now() - timedelta(seconds=1)
session.commit()
result = service.dispatch_due_routes(session)
session.commit()
session.refresh(pending)
self.assertEqual(1, result["delivered"])
self.assertEqual("accepted", pending.status)
self.assertEqual(frozen_target_id, pending.target_postbox_id)
self.assertIsNotNone(pending.target_message_id)
self.assertEqual(3, session.query(PostboxMessage).count())
def test_vacancy_escalation_stops_when_previous_function_is_filled(
self,
) -> None:
with Session(self.engine) as session:
service, source, _target_template = self._create_routing_source(
session,
vacancy_escalation=True,
)
service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=source.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-3",
idempotency_key="routing-vacancy-resolved-1",
subject="No longer escalated",
classification="internal",
expires_at=utc_now() + timedelta(days=7),
),
)
session.commit()
pending = (
session.query(PostboxRoute)
.filter(
PostboxRoute.status
== "pending_vacancy_escalation"
)
.one()
)
pending.execute_after = utc_now() - timedelta(seconds=1)
self.idm.assignments.append(self.assignment)
session.commit()
result = service.dispatch_due_routes(session)
session.commit()
session.refresh(pending)
self.assertEqual(1, result["cancelled"])
self.assertEqual(
"cancelled_vacancy_resolved",
pending.status,
)
self.assertIsNone(pending.target_message_id)
self.assertEqual(2, session.query(PostboxMessage).count())
def test_grouping_update_retains_temporarily_hidden_sources(self) -> None:
child_assignment = OrganizationFunctionAssignmentRef(
id="assignment-child",