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 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, ) -> 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", 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_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) 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", 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()