Files

2519 lines
96 KiB
Python

from __future__ import annotations
import hashlib
import unittest
from dataclasses import replace
from datetime import timedelta
from types import SimpleNamespace
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.events import EventBus, event_bus_context
from govoplan_core.core.identity import IdentityRef
from govoplan_core.core.idm import (
OrganizationFunctionAssignmentRef,
OrganizationFunctionIncumbencyRef,
)
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,
PostboxDeliveryRequest,
PostboxExternalRecipientTokenRef,
PostboxTargetRef,
PostboxWrappedKeyRef,
)
from govoplan_core.db.base import Base
from govoplan_core.security.time import utc_now
from govoplan_postbox.backend.db.models import (
Postbox,
PostboxAccessEvent,
PostboxAddress,
PostboxAttachmentReference,
PostboxBinding,
PostboxDelivery,
PostboxGrouping,
PostboxGroupingSource,
PostboxMessage,
PostboxMessageReceipt,
PostboxParticipant,
PostboxProtectionTransition,
PostboxProtectionTransitionItem,
PostboxRoute,
PostboxTemplate,
PostboxTemplateRevision,
)
from govoplan_postbox.backend.content_protection import PostboxContentProtectionError
from govoplan_postbox.backend.service import PostboxError, PostboxService
POSTBOX_TABLES = (
PostboxTemplate.__table__,
PostboxTemplateRevision.__table__,
PostboxAddress.__table__,
Postbox.__table__,
PostboxBinding.__table__,
PostboxMessage.__table__,
PostboxProtectionTransition.__table__,
PostboxProtectionTransitionItem.__table__,
PostboxParticipant.__table__,
PostboxAttachmentReference.__table__,
PostboxDelivery.__table__,
PostboxRoute.__table__,
PostboxMessageReceipt.__table__,
PostboxGrouping.__table__,
PostboxGroupingSource.__table__,
PostboxAccessEvent.__table__,
)
class FakeIdentityDirectory:
def get_identity(self, identity_id: str):
return IdentityRef(id=identity_id, primary_account_id="account-1")
def identity_for_account(self, account_id: str):
return IdentityRef(
id="identity-1",
primary_account_id=account_id,
account_ids=(account_id,),
)
def identities_for_accounts(self, account_ids):
return tuple(
self.identity_for_account(account_id) for account_id in account_ids
)
def accounts_for_identity(self, identity_id: str):
return ()
class FakeIdmDirectory:
def __init__(self) -> None:
self.assignments: list[OrganizationFunctionAssignmentRef] = []
def get_organization_function_assignment(self, assignment_id: str):
return next(
(
assignment
for assignment in self.assignments
if assignment.id == assignment_id
),
None,
)
def organization_function_assignments_for_identity(
self,
identity_id: str,
*,
tenant_id: str | None = None,
):
return tuple(
assignment
for assignment in self.assignments
if assignment.identity_id == identity_id
and (tenant_id is None or assignment.tenant_id == tenant_id)
and assignment.status == "active"
)
def organization_function_assignments_for_account(
self,
account_id: str,
*,
tenant_id: str | None = None,
):
return tuple(
assignment
for assignment in self.assignments
if assignment.account_id == account_id
and (tenant_id is None or assignment.tenant_id == tenant_id)
and assignment.status == "active"
)
def organization_function_assignments_for_function(
self,
function_id: str,
*,
tenant_id: str | None = None,
):
return tuple(
assignment
for assignment in self.assignments
if assignment.function_id == function_id
and (tenant_id is None or assignment.tenant_id == tenant_id)
and assignment.status == "active"
)
def organization_function_incumbencies(
self,
function_ids,
*,
tenant_id: str,
effective_at=None,
):
del effective_at
return {
function_id: OrganizationFunctionIncumbencyRef(
tenant_id=tenant_id,
function_id=function_id,
assignments=self.organization_function_assignments_for_function(
function_id,
tenant_id=tenant_id,
),
)
for function_id in function_ids
}
class FakeOrganizationDirectory:
def __init__(self) -> None:
self.units = {
"unit-1": OrganizationUnitRef(
id="unit-1",
tenant_id="tenant-1",
slug="district-north",
name="District North",
unit_type_id="district",
),
"unit-child": OrganizationUnitRef(
id="unit-child",
tenant_id="tenant-1",
slug="service-desk",
name="Service Desk",
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(
id="function-1",
tenant_id="tenant-1",
organization_unit_id="unit-1",
slug="case-clerk",
name="Case Clerk",
function_type_id="case-clerk-type",
delegable=True,
),
"function-child": OrganizationFunctionRef(
id="function-child",
tenant_id="tenant-1",
organization_unit_id="unit-child",
slug="case-clerk",
name="Case Clerk",
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)
def organization_units_for_tenant(self, tenant_id: str):
return tuple(
unit for unit in self.units.values() if unit.tenant_id == tenant_id
)
def get_function(self, function_id: str):
return self.functions.get(function_id)
def functions_for_organization_unit(
self,
organization_unit_id: str,
*,
include_subunits: bool = False,
):
return tuple(
function
for function in self.functions.values()
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:
self.requests = []
def enqueue_notification(
self,
session,
request,
*,
enqueue_delivery=True,
):
del session
self.requests.append((request, enqueue_delivery))
return {"id": f"notification-{len(self.requests)}"}
class FailingOnceNotificationDispatch(FakeNotificationDispatch):
def __init__(self) -> None:
super().__init__()
self.fail_next = True
def enqueue_notification(
self,
session,
request,
*,
enqueue_delivery=True,
):
if self.fail_next:
self.fail_next = False
raise RuntimeError("notification provider unavailable")
return super().enqueue_notification(
session,
request,
enqueue_delivery=enqueue_delivery,
)
class PostboxServiceTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine, tables=POSTBOX_TABLES)
self.idm = FakeIdmDirectory()
self.organizations = FakeOrganizationDirectory()
self.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]
)
self.assignment = OrganizationFunctionAssignmentRef(
id="assignment-1",
tenant_id="tenant-1",
identity_id="identity-1",
account_id="account-1",
function_id="function-1",
organization_unit_id="unit-1",
source="direct",
)
self.actor = PostboxActorRef(
account_id="account-1",
identity_id="identity-1",
authorized_actions=frozenset({"discover", "read", "send", "acknowledge"}),
)
def tearDown(self) -> None:
self.engine.dispose()
def _create_exact(
self,
session: Session,
*,
encryption_profile: str = "plaintext_v1",
encryption_vault_id: str | None = None,
protection_policy: dict[str, object] | None = None,
grouping_policy: dict[str, object] | None = None,
) -> Postbox:
return self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="District North / Case Clerk Intake",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
encryption_profile=encryption_profile,
encryption_vault_id=encryption_vault_id,
protection_policy=protection_policy,
grouping_policy=grouping_policy,
)
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,
notifications: FakeNotificationDispatch | None = None,
) -> 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]
notifications=notifications, # 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",
expected_revision=target_template.resource_revision,
)
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",
expected_revision=source_template.resource_revision,
)
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)
session.commit()
denied = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=self.actor,
action="read",
)
self.assertFalse(denied.allowed)
self.assertTrue(denied.vacant)
self.idm.assignments.append(self.assignment)
allowed = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=self.actor,
action="read",
)
self.assertTrue(allowed.allowed)
self.assertFalse(allowed.vacant)
self.assertEqual("assignment-1", allowed.selected_assignment_id)
self.idm.assignments.clear()
revoked = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=self.actor,
action="read",
)
self.assertFalse(revoked.allowed)
def test_acting_assignment_requires_selected_context(self) -> None:
acting = OrganizationFunctionAssignmentRef(
id="acting-assignment",
tenant_id="tenant-1",
identity_id="identity-1",
account_id="account-1",
function_id="function-1",
organization_unit_id="unit-1",
source="acting_for",
delegated_from_assignment_id="source-assignment",
acting_for_account_id="represented-account",
)
self.idm.assignments.append(acting)
with Session(self.engine) as session:
postbox = self._create_exact(session)
session.commit()
denied = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=self.actor,
action="read",
)
allowed = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=PostboxActorRef(
account_id="account-1",
identity_id="identity-1",
selected_assignment_id=acting.id,
acting_for_account_id="represented-account",
authorized_actions=frozenset({"read"}),
),
action="read",
)
self.assertFalse(denied.allowed)
self.assertTrue(allowed.allowed)
self.assertEqual("effective_acting_for_assignment", allowed.reason_code)
def test_classification_clearance_controls_directory_and_delivery(self) -> None:
self.idm.assignments.append(self.assignment)
elevated_actor = PostboxActorRef(
account_id="account-1",
identity_id="identity-1",
authorized_actions=frozenset({"discover", "read", "send", "reply"}),
authorized_classifications=frozenset(
{"public", "internal", "confidential"}
),
)
with Session(self.engine) as session:
postbox = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Confidential intake",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="confidential",
actor_id="admin-1",
)
denied = self.service.explain_access(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
actor=self.actor,
action="read",
)
visible = self.service.list_visible_postboxes(
session,
tenant_id="tenant-1",
actor=elevated_actor,
)
delivered = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="recipient",
producer_resource_id="recipient-1",
idempotency_key="confidential-message",
subject="Confidential notice",
classification="confidential",
),
)
self.assertEqual(
denied.reason_code,
"classification_clearance_missing",
)
self.assertEqual([postbox.id], [item.id for item in visible])
self.assertIsNotNone(
self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=elevated_actor,
)
)
self.assertIsNone(
self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
)
with self.assertRaisesRegex(
PostboxError,
"exceeds the target Postbox classification",
):
self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="recipient",
producer_resource_id="recipient-2",
idempotency_key="restricted-message",
subject="Restricted notice",
classification="restricted",
),
)
def test_template_revision_is_immutable_and_materialization_idempotent(
self,
) -> None:
with Session(self.engine) as session:
template = self.service.create_template(
session,
tenant_id="tenant-1",
slug="case-intake",
name="Case intake",
description=None,
function_type_id="case-clerk-type",
scope_kind="subtree",
scope_id="unit-1",
name_pattern="{unit_name} / {function_name} Intake",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
actor_id="admin-1",
)
self.service.publish_template(
session,
tenant_id="tenant-1",
template_id=template.id,
revision_number=1,
actor_id="admin-1",
expected_revision=template.resource_revision,
)
first = self.service.materialize_template(
session,
tenant_id="tenant-1",
template_id=template.id,
organization_unit_id="unit-child",
function_id="function-child",
context_key="case-42",
actor_id="admin-1",
)
second = self.service.materialize_template(
session,
tenant_id="tenant-1",
template_id=template.id,
organization_unit_id="unit-child",
function_id="function-child",
context_key="case-42",
actor_id="admin-1",
)
revised = self.service.revise_template(
session,
tenant_id="tenant-1",
template_id=template.id,
function_type_id="case-clerk-type",
scope_kind="subtree",
scope_id="unit-1",
name_pattern="{unit_name} / {function_name} Work",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="restricted",
allow_vacant_delivery=True,
actor_id="admin-1",
expected_revision=template.resource_revision,
)
self.assertEqual(first.id, second.id)
self.assertEqual(2, len(revised.revisions))
self.assertEqual(
"{unit_name} / {function_name} Intake",
revised.revisions[0].name_pattern,
)
self.assertEqual(
"{unit_name} / {function_name} Work",
revised.revisions[1].name_pattern,
)
def test_template_preview_is_read_only_and_explains_existing_vacant_and_colliding_targets(
self,
) -> None:
self.idm.assignments = [self.assignment]
with Session(self.engine) as session:
template = self.service.create_template(
session,
tenant_id="tenant-1",
slug="case-intake",
name="Case intake",
description=None,
function_type_id="case-clerk-type",
scope_kind="tenant",
scope_id=None,
name_pattern="{unit_name} / {function_name}",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
actor_id="admin-1",
)
self.service.publish_template(
session,
tenant_id="tenant-1",
template_id=template.id,
revision_number=None,
actor_id="admin-1",
expected_revision=template.resource_revision,
)
self.service.materialize_template(
session,
tenant_id="tenant-1",
template_id=template.id,
organization_unit_id="unit-1",
function_id="function-1",
context_key=None,
actor_id="admin-1",
)
before = session.query(Postbox).count()
preview = self.service.preview_template_targets(
session,
tenant_id="tenant-1",
template_id=template.id,
slug=template.slug,
name=template.name,
description=None,
function_type_id="case-clerk-type",
scope_kind="tenant",
scope_id=None,
scope_structure_id=None,
scope_relation_type_ids=(),
name_pattern="{unit_name} / {function_name}",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
routing_policy={},
encryption_profile="plaintext_v1",
encryption_vault_id=None,
protection_policy={},
context_key=None,
limit=200,
)
self.assertEqual(3, preview["total"])
self.assertEqual(1, preview["existing_count"])
self.assertEqual(2, preview["vacant_count"])
self.assertEqual(before, session.query(Postbox).count())
collision_preview = self.service.preview_template_targets(
session,
tenant_id="tenant-1",
template_id=None,
slug="collision",
name="Collision",
description=None,
function_type_id="case-clerk-type",
scope_kind="tenant",
scope_id=None,
scope_structure_id=None,
scope_relation_type_ids=(),
name_pattern="{unit_name} / {function_name}",
address_pattern="same-address",
classification="internal",
allow_vacant_delivery=True,
routing_policy={},
encryption_profile="plaintext_v1",
encryption_vault_id=None,
protection_policy={},
context_key=None,
limit=200,
)
self.assertEqual(3, collision_preview["blocked_count"])
self.assertIn(
"duplicate_generated_address",
collision_preview["diagnostics"],
)
self.organizations.duplicate_parent_match = True
hierarchy_preview = self.service.preview_template_targets(
session,
tenant_id="tenant-1",
template_id=None,
slug="hierarchy",
name="Hierarchy",
description=None,
function_type_id="case-clerk-type",
scope_kind="subtree",
scope_id="unit-child",
scope_structure_id="structure-1",
scope_relation_type_ids=("relation-type-1",),
name_pattern="{unit_name} / {function_name}",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
routing_policy={},
encryption_profile="plaintext_v1",
encryption_vault_id=None,
protection_policy={},
context_key=None,
limit=200,
)
self.assertIn(
"ambiguous_scope_paths",
hierarchy_preview["diagnostics"],
)
self.assertEqual(
"descendants",
self.organizations.last_hierarchy_request["direction"],
)
def test_delivery_catalog_exposes_exact_and_derived_target_choices(
self,
) -> None:
with Session(self.engine) as session:
exact = self._create_exact(session)
template = self.service.create_template(
session,
tenant_id="tenant-1",
slug="case-intake",
name="Case intake",
description="Functional intake",
function_type_id="case-clerk-type",
scope_kind="subtree",
scope_id="unit-1",
name_pattern="{unit_name} / {function_name} Intake",
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
classification="internal",
allow_vacant_delivery=True,
actor_id="admin-1",
)
self.service.publish_template(
session,
tenant_id="tenant-1",
template_id=template.id,
revision_number=None,
actor_id="admin-1",
expected_revision=template.resource_revision,
)
session.commit()
catalog = self.service.delivery_catalog(
session,
tenant_id="tenant-1",
)
self.assertEqual([exact.id], [item.id for item in catalog.postboxes])
self.assertEqual(
["case-intake"],
[item.slug for item in catalog.templates],
)
north = next(
unit for unit in catalog.organization_units if unit.id == "unit-1"
)
self.assertEqual(
["function-1"],
[function.id for function in north.functions],
)
def test_delivery_is_idempotent_and_receipts_are_per_account(self) -> None:
self.idm.assignments.append(self.assignment)
notifications = FakeNotificationDispatch()
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]
notifications=notifications, # type: ignore[arg-type]
)
events = []
event_bus = EventBus()
event_bus.subscribe("*", events.append)
with Session(self.engine) as session:
postbox = service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="District North / Case Clerk Intake",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
)
request = PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-1",
idempotency_key="campaign-1:recipient-1:postbox",
subject="Permit decision",
body_text="The decision is available.",
expires_at=utc_now() + timedelta(days=30),
)
with event_bus_context(event_bus):
first = service.deliver(session, request)
second = service.deliver(session, request)
self.assertEqual(
1,
service.count_messages(
session,
tenant_id="tenant-1",
postbox_ids=(postbox.id,),
actor=self.actor,
query="permit",
state="unread",
),
)
marked = service.mark_message(
session,
tenant_id="tenant-1",
message_id=first.message_id,
actor=self.actor,
state="acknowledged",
)
service.mark_message(
session,
tenant_id="tenant-1",
message_id=first.message_id,
actor=self.actor,
state="acknowledged",
)
receipt_summary = service.delivery_receipt_summaries(
session,
tenant_id="tenant-1",
producer_module="campaigns",
delivery_ids=(first.delivery_id,),
)[first.delivery_id]
session.commit()
self.assertFalse(first.duplicate)
self.assertTrue(second.duplicate)
self.assertEqual(first.message_id, second.message_id)
self.assertIsNotNone(marked.read_at)
self.assertIsNotNone(marked.acknowledged_at)
self.assertTrue(receipt_summary.currently_readable)
self.assertEqual(1, receipt_summary.current_holder_count)
self.assertEqual(1, receipt_summary.read_receipt_count)
self.assertEqual(1, receipt_summary.acknowledged_receipt_count)
self.assertEqual(0, receipt_summary.withdrawn_message_count)
self.assertEqual(
{},
service.delivery_receipt_summaries(
session,
tenant_id="tenant-1",
producer_module="another-module",
delivery_ids=(first.delivery_id,),
),
)
self.assertEqual(
1,
session.query(PostboxMessage).count(),
)
self.assertEqual(
1,
session.query(PostboxDelivery).count(),
)
self.assertEqual(
1,
session.query(PostboxMessageReceipt).count(),
)
self.assertEqual(
(),
service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(postbox.id,),
actor=self.actor,
query="not present",
),
)
self.assertEqual(
1,
service.count_messages(
session,
tenant_id="tenant-1",
postbox_ids=(postbox.id,),
actor=self.actor,
state="acknowledged",
),
)
self.assertEqual(1, len(notifications.requests))
notification, enqueue_delivery = notifications.requests[0]
self.assertEqual("account-1", notification.recipient_id)
self.assertEqual("account", notification.recipient_type)
self.assertEqual(
f"/postbox?message={first.message_id}",
notification.action_url,
)
self.assertFalse(enqueue_delivery)
self.assertEqual(
[
"postbox.delivery.accepted.v1",
"postbox.message.acknowledged.v1",
],
[event.type for event in events],
)
def test_notification_lifecycle_reconciles_assignment_churn_and_retries(
self,
) -> None:
self.idm.assignments.append(self.assignment)
notifications = FailingOnceNotificationDispatch()
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]
notifications=notifications, # type: ignore[arg-type]
)
event_bus = EventBus()
events = []
event_bus.subscribe("*", events.append)
with Session(self.engine) as session, event_bus_context(event_bus):
postbox = service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="District North / Lifecycle notifications",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
)
session.commit()
baseline = service.reconcile_notification_lifecycle(
session,
tenant_id="tenant-1",
)
session.commit()
self.assertEqual(1, baseline["scanned"])
self.assertEqual(0, baseline["changed"])
baseline_event_count = len(events)
delegated = replace(
self.assignment,
id="assignment-delegated",
identity_id="identity-2",
account_id="account-2",
source="delegated",
delegated_from_assignment_id=self.assignment.id,
)
overlapping_delegation = replace(
delegated,
id="assignment-delegated-overlap",
)
self.idm.assignments.extend((delegated, overlapping_delegation))
added = service.reconcile_notification_lifecycle(
session,
tenant_id="tenant-1",
)
session.commit()
self.assertEqual(1, added["changed"])
self.assertEqual(3, added["events"])
self.assertEqual(0, added["notifications"])
self.assertEqual(1, added["notification_failures"])
retried = service.reconcile_notification_lifecycle(
session,
tenant_id="tenant-1",
)
session.commit()
self.assertEqual(0, retried["changed"])
self.assertEqual(0, retried["events"])
self.assertEqual(1, retried["notifications"])
self.assertEqual(1, len(notifications.requests))
request, enqueue_delivery = notifications.requests[0]
self.assertEqual("postbox.delegation.newly_visible.v1", request.event_kind)
self.assertEqual("account-2", request.recipient_id)
self.assertEqual(f"/postbox?postbox={postbox.id}", request.action_url)
self.assertFalse(enqueue_delivery)
self.idm.assignments.clear()
vacant = service.reconcile_notification_lifecycle(
session,
tenant_id="tenant-1",
)
session.commit()
self.assertEqual(1, vacant["changed"])
self.assertEqual(5, vacant["events"])
replacement = replace(
self.assignment,
id="assignment-replacement",
identity_id="identity-3",
account_id="account-3",
)
self.idm.assignments.append(replacement)
filled = service.reconcile_notification_lifecycle(
session,
tenant_id="tenant-1",
)
session.commit()
self.assertEqual(1, filled["changed"])
self.assertEqual(3, filled["events"])
self.assertEqual(1, filled["notifications"])
lifecycle_events = events[baseline_event_count:]
self.assertEqual(
{
"postbox.assignment.reassigned.v1",
"postbox.assignment.newly_visible.v1",
"postbox.assignment.visibility_withdrawn.v1",
"postbox.delegation.newly_visible.v1",
"postbox.vacancy.detected.v1",
"postbox.vacancy.resolved.v1",
},
{event.type for event in lifecycle_events},
)
for event in lifecycle_events:
serialized = repr(dict(event.payload)).casefold()
self.assertNotIn("body", serialized)
self.assertNotIn("attachment", serialized)
self.assertNotIn("subject", serialized)
def test_action_required_delivery_uses_metadata_only_notification(self) -> None:
self.idm.assignments.append(self.assignment)
notifications = FakeNotificationDispatch()
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]
notifications=notifications, # type: ignore[arg-type]
)
events = []
event_bus = EventBus()
event_bus.subscribe("*", events.append)
with Session(self.engine) as session, event_bus_context(event_bus):
postbox = service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="District North / Required action",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
)
delivered = service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="tests",
producer_resource_type="required_action",
producer_resource_id="action-1",
idempotency_key="required-action-1",
subject="Sensitive subject is not copied",
body_text="Sensitive body is not copied",
action_required=True,
),
)
session.commit()
self.assertIn(
"postbox.message.action_required.v1",
[event.type for event in events],
)
notification, _enqueue_delivery = notifications.requests[0]
self.assertEqual(
"postbox.message.action_required.v1",
notification.event_kind,
)
self.assertEqual(4, notification.priority)
self.assertEqual(delivered.message_id, notification.payload["message_id"])
self.assertNotIn("Sensitive", notification.body_text or "")
self.assertNotIn("Sensitive", notification.subject or "")
def test_withdrawn_and_expired_messages_keep_metadata_but_hide_content(
self,
) -> None:
self.idm.assignments.append(self.assignment)
with Session(self.engine) as session:
postbox = self._create_exact(session)
withdrawn = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-withdrawn",
idempotency_key="withdrawn-message",
subject="Withdrawn decision",
body_text="Sensitive withdrawn content",
),
)
expired = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-expired",
idempotency_key="expired-message",
subject="Expired decision",
body_text="Sensitive expired content",
expires_at=utc_now() - timedelta(minutes=1),
),
)
withdrawn_model = session.get(PostboxMessage, withdrawn.message_id)
assert withdrawn_model is not None
withdrawn_model.withdrawn_at = utc_now()
session.flush()
withdrawn_ref = self.service.get_message(
session,
tenant_id="tenant-1",
message_id=withdrawn.message_id,
actor=self.actor,
)
expired_ref = self.service.get_message(
session,
tenant_id="tenant-1",
message_id=expired.message_id,
actor=self.actor,
)
assert withdrawn_ref is not None
assert expired_ref is not None
self.assertEqual("withdrawn", withdrawn_ref.availability)
self.assertIsNone(withdrawn_ref.body_text)
self.assertEqual((), withdrawn_ref.attachments)
self.assertEqual("expired", expired_ref.availability)
self.assertIsNone(expired_ref.body_text)
with self.assertRaisesRegex(PostboxError, "withdrawn"):
self.service.mark_message(
session,
tenant_id="tenant-1",
message_id=withdrawn.message_id,
actor=self.actor,
state="read",
)
with self.assertRaisesRegex(PostboxError, "expired"):
self.service.mark_message(
session,
tenant_id="tenant-1",
message_id=expired.message_id,
actor=self.actor,
state="acknowledged",
)
summaries = self.service.delivery_receipt_summaries(
session,
tenant_id="tenant-1",
producer_module="campaigns",
delivery_ids=(withdrawn.delivery_id, expired.delivery_id),
)
self.assertFalse(summaries[withdrawn.delivery_id].currently_readable)
self.assertEqual(
1,
summaries[withdrawn.delivery_id].withdrawn_message_count,
)
self.assertFalse(summaries[expired.delivery_id].currently_readable)
self.assertEqual(1, summaries[expired.delivery_id].expired_message_count)
def test_encrypted_envelope_and_external_grant_state_cross_capability(self) -> None:
self.idm.assignments.append(self.assignment)
expires_at = utc_now() + timedelta(days=1)
with Session(self.engine) as session:
postbox = self._create_exact(
session,
encryption_profile="external_e2ee_v1",
protection_policy={"external_recipient_assurance": "email_otp"},
)
delivered = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-encrypted",
idempotency_key="encrypted-message",
subject="Encrypted decision",
ciphertext_ref="files:ciphertext-1",
signed_manifest_ref="files:manifest-1",
wrapped_keys=(
PostboxWrappedKeyRef(
recipient_type="function_postbox",
recipient_id=postbox.id,
key_epoch=postbox.key_epoch,
wrapped_key_ref="trust:wrapped-key-1",
algorithm="HPKE-v1",
),
),
external_recipient_tokens=(
PostboxExternalRecipientTokenRef(
token_id="grant-1",
state="available",
expires_at=expires_at,
one_time=True,
assurance_profile="email-otp",
),
),
metadata={"content_digest": "sha256:" + "a" * 64},
),
)
message = self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
assert message is not None
self.assertEqual("files:ciphertext-1", message.ciphertext_ref)
self.assertEqual(
"trust:wrapped-key-1",
message.wrapped_keys[0].wrapped_key_ref,
)
self.assertEqual("grant-1", message.external_recipient_tokens[0].token_id)
self.assertEqual("available", message.external_recipient_tokens[0].state)
assert message.external_recipient_tokens[0].expires_at is not None
self.assertEqual(
expires_at.replace(microsecond=0),
message.external_recipient_tokens[0].expires_at.replace(microsecond=0),
)
def test_server_envelope_body_is_not_persisted_in_plaintext_and_fails_closed(
self,
) -> None:
self.idm.assignments.append(self.assignment)
protected = SimpleNamespace(
ciphertext=b"encrypted-message-body",
envelope=SimpleNamespace(
envelope_id="envelope-1",
ciphertext_ref="postbox-db://message/body",
algorithm_suite="AES-256-GCM",
wrapped_key_refs=("wrapped-key-1",),
),
)
with Session(self.engine) as session:
postbox = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Protected intake",
organization_unit_id="unit-1",
function_id="function-1",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
encryption_profile="server_envelope_v1",
encryption_vault_id="vault-1",
)
with patch(
"govoplan_postbox.backend.content_protection.protect_message_body",
return_value=protected,
):
delivered = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-protected",
idempotency_key="protected-message",
subject="Protected notice",
body_text="clear message body",
),
)
stored = session.get(PostboxMessage, delivered.message_id)
assert stored is not None
self.assertIsNone(stored.body_text)
self.assertEqual(b"encrypted-message-body", stored.body_ciphertext)
self.assertEqual("envelope-1", stored.encryption_envelope_id)
with patch(
"govoplan_postbox.backend.content_protection.unprotect_message_body",
return_value="clear message body",
):
opened = self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
assert opened is not None
self.assertEqual("clear message body", opened.body_text)
with patch(
"govoplan_postbox.backend.content_protection.unprotect_message_body",
side_effect=PostboxContentProtectionError("provider unavailable"),
):
with self.assertRaisesRegex(
PostboxError,
"provider unavailable",
):
self.service.get_message(
session,
tenant_id="tenant-1",
message_id=delivered.message_id,
actor=self.actor,
)
def test_future_only_protection_transition_changes_new_message_policy(self) -> None:
with Session(self.engine) as session:
postbox = self._create_exact(
session,
protection_policy={
"handover_authority": "user_consent",
"handover_quorum": 1,
},
)
transition = self.service.create_protection_transition(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
idempotency_key="future-e2ee",
expected_revision=postbox.resource_revision,
target_profile="external_e2ee_v1",
target_vault_id=None,
history_mode="future_only",
authority_mode="user_consent",
required_quorum=1,
user_consent_refs=("consent:user-1",),
institutional_authorization_refs=(),
reason="Use client-held keys for future messages.",
actor_id="admin-1",
)
self.assertEqual("completed", transition.state)
self.assertEqual(0, transition.message_count)
self.assertEqual("external_e2ee_v1", postbox.encryption_profile)
self.assertEqual(2, postbox.key_epoch)
def test_client_transform_completes_plaintext_to_e2ee_history(self) -> None:
self.idm.assignments.append(self.assignment)
plaintext = "A governed message"
digest = "sha256:" + hashlib.sha256(plaintext.encode()).hexdigest()
with Session(self.engine) as session:
postbox = self._create_exact(
session,
protection_policy={
"handover_authority": "user_consent",
"handover_quorum": 1,
},
)
delivered = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-1",
idempotency_key="plaintext-history",
subject="Governed history",
body_text=plaintext,
),
)
transition = self.service.create_protection_transition(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
idempotency_key="migrate-e2ee",
expected_revision=postbox.resource_revision,
target_profile="external_e2ee_v1",
target_vault_id=None,
history_mode="migrate_history",
authority_mode="user_consent",
required_quorum=1,
user_consent_refs=("consent:user-1",),
institutional_authorization_refs=(),
reason="Move retained content to client-held keys.",
actor_id="admin-1",
)
self.assertEqual("awaiting_client", transition.state)
self.assertEqual(1, transition.message_count)
completed = self.service.apply_client_protection_transform(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
transition_id=transition.id,
message_id=delivered.message_id,
expected_revision=transition.resource_revision,
plaintext=None,
ciphertext_ref="files:ciphertext-transition-1",
signed_manifest_ref="files:manifest-transition-1",
wrapped_keys=(
PostboxWrappedKeyRef(
recipient_type="function_postbox",
recipient_id=postbox.id,
key_epoch=postbox.key_epoch,
wrapped_key_ref="trust:wrapped-transition-1",
algorithm="HPKE-v1",
),
),
content_digest=digest,
transformation_evidence_ref="client:evidence-1",
actor_id="admin-1",
)
self.assertEqual("completed", completed.state)
self.assertEqual(1, completed.completed_count)
stored = session.get(PostboxMessage, delivered.message_id)
assert stored is not None
self.assertIsNone(stored.body_text)
self.assertEqual("external_e2ee_v1", stored.encryption_profile)
self.assertEqual("files:ciphertext-transition-1", stored.ciphertext_ref)
self.assertEqual(digest, stored.metadata_["content_digest"])
def test_plaintext_history_migrates_to_managed_envelopes_automatically(
self,
) -> None:
protected = SimpleNamespace(
ciphertext=b"managed-history",
envelope=SimpleNamespace(
envelope_id="transition-envelope-1",
ciphertext_ref="postbox-db://messages/history/body",
algorithm_suite="AES-256-GCM",
wrapped_key_refs=("wrapped-history-1",),
vault_id="vault-1",
key_version=4,
),
)
with Session(self.engine) as session:
postbox = self._create_exact(session)
delivered = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-managed-history",
idempotency_key="managed-history",
subject="Managed history",
body_text="Move this content",
),
)
with patch(
"govoplan_postbox.backend.content_protection.protect_message_body",
return_value=protected,
):
transition = self.service.create_protection_transition(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
idempotency_key="migrate-managed",
expected_revision=postbox.resource_revision,
target_profile="server_envelope_v1",
target_vault_id="vault-1",
history_mode="migrate_history",
authority_mode="dual_control",
required_quorum=2,
user_consent_refs=("consent:incumbent-1",),
institutional_authorization_refs=("approval:key-holder-1",),
reason="Adopt the managed standard.",
actor_id="admin-1",
)
self.assertEqual("completed", transition.state)
stored = session.get(PostboxMessage, delivered.message_id)
assert stored is not None
self.assertIsNone(stored.body_text)
self.assertEqual(b"managed-history", stored.body_ciphertext)
self.assertEqual("transition-envelope-1", stored.encryption_envelope_id)
self.assertEqual("server_envelope_v1", stored.encryption_profile)
self.assertEqual("vault-1", stored.wrapped_keys[0]["recipient_id"])
def test_new_incumbent_history_policy_filters_lists_counts_and_direct_reads(
self,
) -> None:
boundary = utc_now() - timedelta(days=1)
self.idm.assignments.append(replace(self.assignment, valid_from=boundary))
with Session(self.engine) as session:
postbox = self._create_exact(
session,
protection_policy={"new_incumbent_history": "since_assignment"},
)
old_delivery = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-old",
idempotency_key="history-old",
subject="Before assignment",
body_text="Old content",
),
)
recent_delivery = self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-recent",
idempotency_key="history-recent",
subject="After assignment",
body_text="Recent content",
),
)
old = session.get(PostboxMessage, old_delivery.message_id)
recent = session.get(PostboxMessage, recent_delivery.message_id)
assert old is not None and recent is not None
old.delivered_at = boundary - timedelta(hours=1)
recent.delivered_at = boundary + timedelta(hours=1)
session.flush()
messages = self.service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(postbox.id,),
actor=self.actor,
)
self.assertEqual([recent.id], [message.id for message in messages])
self.assertEqual(
1,
self.service.count_messages(
session,
tenant_id="tenant-1",
postbox_ids=(postbox.id,),
actor=self.actor,
),
)
self.assertIsNone(
self.service.get_message(
session,
tenant_id="tenant-1",
message_id=old.id,
actor=self.actor,
)
)
self.assertFalse(
self.service.can_read_message(
session,
tenant_id="tenant-1",
message_id=old.id,
actor=self.actor,
)
)
def test_leaving_e2ee_history_requires_user_consent_authority(self) -> None:
with Session(self.engine) as session:
postbox = self._create_exact(
session,
encryption_profile="external_e2ee_v1",
)
with self.assertRaisesRegex(PostboxError, "user consent"):
self.service.create_protection_transition(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
idempotency_key="leave-e2ee-without-user",
expected_revision=postbox.resource_revision,
target_profile="plaintext_v1",
target_vault_id=None,
history_mode="future_only",
authority_mode="institutional_key_holders",
required_quorum=1,
user_consent_refs=(),
institutional_authorization_refs=("approval:key-holder-1",),
reason="Leave E2EE without holder consent.",
actor_id="admin-1",
)
def test_protection_policy_update_is_revisioned_and_enforces_assurance(
self,
) -> None:
with Session(self.engine) as session:
postbox = self._create_exact(session)
original_revision = postbox.resource_revision
updated = self.service.update_protection_policy(
session,
tenant_id="tenant-1",
postbox_id=postbox.id,
protection_policy={
"new_incumbent_history": "all_retained",
"external_recipient_assurance": "disabled",
},
actor_id="admin-1",
expected_revision=original_revision,
)
self.assertEqual(original_revision + 1, updated.resource_revision)
self.assertEqual(
"disabled",
updated.settings["protection_policy"]["external_recipient_assurance"],
)
self.assertEqual(
"rewrap",
updated.settings["protection_policy"]["ordinary_rotation"],
)
with self.assertRaisesRegex(PostboxError, "does not permit"):
self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="campaigns",
producer_resource_type="campaign_recipient",
producer_resource_id="recipient-external-disabled",
idempotency_key="external-disabled",
subject="External retrieval",
body_text="Content",
external_recipient_tokens=(
PostboxExternalRecipientTokenRef(
token_id="grant-disabled",
state="available",
one_time=True,
assurance_profile="strong_identity",
),
),
),
)
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]
)
summary = service.delivery_receipt_summaries(
session,
tenant_id="tenant-1",
producer_module="campaigns",
delivery_ids=(result.delivery_id,),
)[result.delivery_id]
self.assertEqual(2, summary.message_count)
self.assertEqual(1, summary.routed_message_count)
self.assertEqual(1, summary.read_receipt_count)
self.assertEqual({"accepted": 1}, summary.route_status_counts)
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)
notifications = FakeNotificationDispatch()
events = []
event_bus = EventBus()
event_bus.subscribe("*", events.append)
with Session(self.engine) as session, event_bus_context(event_bus):
service, source, _target_template = self._create_routing_source(
session,
vacancy_escalation=True,
notifications=notifications,
)
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())
self.assertIn(
"postbox.escalation.due.v1",
[event.type for event in events],
)
escalation_notification = next(
request
for request, _enqueue_delivery in notifications.requests
if request.event_kind == "postbox.escalation.due.v1"
)
self.assertEqual("account-top", escalation_notification.recipient_id)
self.assertEqual(4, escalation_notification.priority)
self.assertNotIn("Escalated decision", escalation_notification.body_text)
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",
tenant_id="tenant-1",
identity_id="identity-1",
account_id="account-1",
function_id="function-child",
organization_unit_id="unit-child",
source="direct",
)
self.idm.assignments.extend((self.assignment, child_assignment))
with Session(self.engine) as session:
parent = self._create_exact(session)
child = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Service Desk / Case Clerk Intake",
organization_unit_id="unit-child",
function_id="function-child",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
)
grouping = self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=self.actor,
grouping_id=None,
name="Assigned work",
is_default=True,
postbox_ids=(parent.id, child.id),
)
session.commit()
grouping_id = grouping.id
self.idm.assignments.remove(child_assignment)
updated = self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=self.actor,
grouping_id=grouping_id,
expected_revision=grouping.resource_revision,
name="Assigned work",
is_default=True,
postbox_ids=(parent.id,),
)
session.commit()
self.assertEqual(
(parent.id, child.id),
tuple(source.postbox_id for source in updated.sources),
)
self.idm.assignments.append(child_assignment)
visible_ids = {
item.id
for item in self.service.list_visible_postboxes(
session,
tenant_id="tenant-1",
actor=self.actor,
)
}
self.assertEqual({parent.id, child.id}, visible_ids)
def test_grouping_policy_enforces_function_and_classification_separation(
self,
) -> None:
child_assignment = OrganizationFunctionAssignmentRef(
id="assignment-child",
tenant_id="tenant-1",
identity_id="identity-1",
account_id="account-1",
function_id="function-child",
organization_unit_id="unit-child",
source="delegated",
delegated_from_assignment_id="assignment-1",
)
self.idm.assignments.extend((self.assignment, child_assignment))
privileged_actor = replace(
self.actor,
authorized_classifications=frozenset(
{"public", "internal", "confidential"}
),
)
with Session(self.engine) as session:
parent = self._create_exact(
session,
grouping_policy={
"mode": "same_classification",
"reason": "Confidential responsibilities remain partitioned.",
},
)
child = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Service Desk / Delegated complaints",
organization_unit_id="unit-child",
function_id="function-child",
address_key=None,
description=None,
classification="confidential",
actor_id="admin-1",
)
with self.assertRaisesRegex(
PostboxError,
"classification_separation_required",
):
self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=privileged_actor,
grouping_id=None,
name="Mixed classification",
is_default=False,
postbox_ids=(parent.id, child.id),
)
grouping = self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=privileged_actor,
grouping_id=None,
name="Same responsibility",
is_default=True,
postbox_ids=(parent.id,),
)
session.commit()
self.service.update_grouping_policy(
session,
tenant_id="tenant-1",
postbox_id=parent.id,
grouping_policy={
"mode": "separate",
"reason": "This function must remain a dedicated inbox.",
},
actor_id="admin-1",
expected_revision=parent.resource_revision,
)
session.commit()
with self.assertRaisesRegex(PostboxError, "source_requires_separation"):
self.service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(parent.id, child.id),
actor=privileged_actor,
)
self.assertEqual(
(parent.id,),
tuple(source.postbox_id for source in grouping.sources),
)
def test_grouping_pagination_is_stable_and_delegation_expiry_hides_source(
self,
) -> None:
delegated = OrganizationFunctionAssignmentRef(
id="assignment-child",
tenant_id="tenant-1",
identity_id="identity-1",
account_id="account-1",
function_id="function-child",
organization_unit_id="unit-child",
source="delegated",
delegated_from_assignment_id="assignment-1",
)
self.idm.assignments.extend((self.assignment, delegated))
with Session(self.engine) as session:
parent = self._create_exact(session)
child = self.service.create_exact_postbox(
session,
tenant_id="tenant-1",
name="Service Desk / Delegated intake",
organization_unit_id="unit-child",
function_id="function-child",
address_key=None,
description=None,
classification="internal",
actor_id="admin-1",
)
grouping = self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=self.actor,
grouping_id=None,
name="Delegated work",
is_default=True,
postbox_ids=(parent.id, child.id),
)
focused_grouping = self.service.save_grouping(
session,
tenant_id="tenant-1",
actor=self.actor,
grouping_id=None,
name="Delegated intake only",
is_default=False,
postbox_ids=(child.id,),
)
self.assertEqual(
(child.id,),
tuple(source.postbox_id for source in focused_grouping.sources),
)
for index, postbox in enumerate((parent, child, parent, child), start=1):
self.service.deliver(
session,
PostboxDeliveryRequest(
tenant_id="tenant-1",
target=PostboxTargetRef(postbox_id=postbox.id),
producer_module="tests",
producer_resource_type="stable_page",
producer_resource_id=str(index),
idempotency_key=f"stable-page-{index}",
subject=f"Message {index}",
classification="internal",
),
)
boundary = utc_now()
session.query(PostboxMessage).update(
{PostboxMessage.delivered_at: boundary}
)
session.commit()
first = self.service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(parent.id, child.id),
actor=self.actor,
limit=2,
offset=0,
)
second = self.service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(parent.id, child.id),
actor=self.actor,
limit=2,
offset=2,
)
repeated = self.service.list_messages(
session,
tenant_id="tenant-1",
postbox_ids=(parent.id, child.id),
actor=self.actor,
limit=2,
offset=0,
)
self.assertEqual(
[item.id for item in first], [item.id for item in repeated]
)
self.assertFalse({item.id for item in first} & {item.id for item in second})
self.idm.assignments = [
self.assignment,
replace(delegated, status="expired"),
]
visible_ids = {
item.id
for item in self.service.list_visible_postboxes(
session,
tenant_id="tenant-1",
actor=self.actor,
)
}
self.assertEqual({parent.id}, visible_ids)
self.assertEqual(
(parent.id, child.id),
tuple(source.postbox_id for source in grouping.sources),
)
if __name__ == "__main__":
unittest.main()