564 lines
20 KiB
Python
564 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.identity import IdentityRef
|
|
from govoplan_core.core.idm import (
|
|
OrganizationFunctionAssignmentRef,
|
|
OrganizationFunctionIncumbencyRef,
|
|
)
|
|
from govoplan_core.core.organizations import (
|
|
OrganizationFunctionRef,
|
|
OrganizationUnitRef,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_postbox.backend.db.models import (
|
|
Postbox,
|
|
PostboxAccessEvent,
|
|
PostboxAddress,
|
|
PostboxAttachmentReference,
|
|
PostboxBinding,
|
|
PostboxDelivery,
|
|
PostboxGrouping,
|
|
PostboxGroupingSource,
|
|
PostboxMessage,
|
|
PostboxMessageReceipt,
|
|
PostboxParticipant,
|
|
PostboxRoute,
|
|
PostboxTemplate,
|
|
PostboxTemplateRevision,
|
|
)
|
|
from govoplan_postbox.backend.router import router
|
|
from govoplan_postbox.backend.portal_projection import PortalProjection
|
|
from govoplan_postbox.backend.service import PostboxService
|
|
|
|
|
|
TABLES = (
|
|
PostboxTemplate.__table__,
|
|
PostboxTemplateRevision.__table__,
|
|
PostboxAddress.__table__,
|
|
Postbox.__table__,
|
|
PostboxBinding.__table__,
|
|
PostboxMessage.__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)
|
|
|
|
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, assignment: OrganizationFunctionAssignmentRef) -> None:
|
|
self.assignment = assignment
|
|
|
|
def get_organization_function_assignment(self, assignment_id: str):
|
|
return self.assignment if assignment_id == self.assignment.id else None
|
|
|
|
def organization_function_assignments_for_identity(
|
|
self,
|
|
identity_id: str,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
):
|
|
return (self.assignment,) if identity_id == self.assignment.identity_id else ()
|
|
|
|
def organization_function_assignments_for_account(
|
|
self,
|
|
account_id: str,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
):
|
|
return (self.assignment,) if account_id == self.assignment.account_id else ()
|
|
|
|
def organization_function_assignments_for_function(
|
|
self,
|
|
function_id: str,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
):
|
|
return (self.assignment,) if function_id == self.assignment.function_id else ()
|
|
|
|
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:
|
|
unit = OrganizationUnitRef(
|
|
id="unit-1",
|
|
tenant_id="tenant-1",
|
|
slug="district",
|
|
name="District",
|
|
)
|
|
function = OrganizationFunctionRef(
|
|
id="function-1",
|
|
tenant_id="tenant-1",
|
|
organization_unit_id="unit-1",
|
|
slug="clerk",
|
|
name="Clerk",
|
|
function_type_id="clerk-type",
|
|
)
|
|
|
|
def get_organization_unit(self, organization_unit_id: str):
|
|
return self.unit if organization_unit_id == self.unit.id else None
|
|
|
|
def organization_units_for_tenant(self, tenant_id: str):
|
|
return (self.unit,) if tenant_id == self.unit.tenant_id else ()
|
|
|
|
def get_function(self, function_id: str):
|
|
return self.function if function_id == self.function.id else None
|
|
|
|
def functions_for_organization_unit(
|
|
self,
|
|
organization_unit_id: str,
|
|
*,
|
|
include_subunits: bool = False,
|
|
):
|
|
return (self.function,) if organization_unit_id == self.unit.id else ()
|
|
|
|
|
|
class PostboxRouterTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(self.engine, tables=TABLES)
|
|
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",
|
|
)
|
|
idm = FakeIdmDirectory(assignment)
|
|
self.service = PostboxService(
|
|
identities=FakeIdentityDirectory(), # type: ignore[arg-type]
|
|
idm=idm, # type: ignore[arg-type]
|
|
incumbencies=idm, # type: ignore[arg-type]
|
|
organizations=FakeOrganizationDirectory(), # type: ignore[arg-type]
|
|
)
|
|
with Session(self.engine) as session:
|
|
self.postbox = self.service.create_exact_postbox(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
name="District / Clerk",
|
|
organization_unit_id="unit-1",
|
|
function_id="function-1",
|
|
address_key=None,
|
|
description=None,
|
|
classification="internal",
|
|
actor_id="account-1",
|
|
)
|
|
session.commit()
|
|
self.postbox_id = self.postbox.id
|
|
|
|
principal = ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
identity_id="identity-1",
|
|
scopes=frozenset(
|
|
{
|
|
"postbox:postbox:read",
|
|
"postbox:message:write",
|
|
"postbox:message:reply",
|
|
"postbox:message:acknowledge",
|
|
"postbox:delivery:write",
|
|
"postbox:binding:admin",
|
|
"postbox:template:admin",
|
|
}
|
|
),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
self.principal = principal
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
|
|
def session_dependency():
|
|
with Session(self.engine) as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = session_dependency
|
|
app.dependency_overrides[get_api_principal] = lambda: principal
|
|
self.patch = patch(
|
|
"govoplan_postbox.backend.router.get_service",
|
|
return_value=self.service,
|
|
)
|
|
self.patch.start()
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self) -> None:
|
|
self.client.close()
|
|
self.patch.stop()
|
|
self.engine.dispose()
|
|
|
|
def test_assignment_context_override_must_belong_to_principal(self) -> None:
|
|
response = self.client.get(
|
|
"/api/v1/postbox/directory",
|
|
params={"assignment_context_id": "assignment-not-granted"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertIn("not active for this principal", response.text)
|
|
|
|
def test_portal_projection_requires_explicit_visibility_and_keeps_postbox_access(self) -> None:
|
|
projection = PortalProjection()
|
|
with Session(self.engine) as session, patch(
|
|
"govoplan_postbox.backend.portal_projection.get_service",
|
|
return_value=self.service,
|
|
):
|
|
self.assertEqual(
|
|
(),
|
|
projection.list_portal_entries(
|
|
session,
|
|
self.principal,
|
|
tenant_id="tenant-1",
|
|
),
|
|
)
|
|
postbox = session.get(Postbox, self.postbox_id)
|
|
assert postbox is not None
|
|
postbox.settings = {**postbox.settings, "portal_visible": True}
|
|
session.flush()
|
|
|
|
entries = projection.list_portal_entries(
|
|
session,
|
|
self.principal,
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
self.assertEqual(1, len(entries))
|
|
self.assertEqual(self.postbox_id, entries[0].postbox.id)
|
|
self.assertEqual(f"/postbox?postbox={self.postbox_id}", entries[0].route_path)
|
|
|
|
def test_template_impact_preview_is_available_without_writes(self) -> None:
|
|
with Session(self.engine) as session:
|
|
before = session.query(Postbox).count()
|
|
response = self.client.post(
|
|
"/api/v1/postbox/admin/templates/preview",
|
|
json={
|
|
"slug": "case-intake",
|
|
"name": "Case intake",
|
|
"scope_kind": "tenant",
|
|
"name_pattern": "{unit_name} / {function_name}",
|
|
"address_pattern": "{template_slug}.{unit_slug}.{function_slug}",
|
|
"classification": "internal",
|
|
},
|
|
)
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertEqual(1, response.json()["total"])
|
|
self.assertEqual(1, response.json()["ready_count"])
|
|
with Session(self.engine) as session:
|
|
self.assertEqual(before, session.query(Postbox).count())
|
|
|
|
def test_legacy_subtree_template_remains_readable_but_cannot_be_created_by_api(
|
|
self,
|
|
) -> None:
|
|
with Session(self.engine) as session:
|
|
self.service.create_template(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
slug="legacy-subtree",
|
|
name="Legacy subtree",
|
|
description=None,
|
|
function_type_id="clerk-type",
|
|
scope_kind="subtree",
|
|
scope_id="unit-1",
|
|
name_pattern="{unit_name} / {function_name}",
|
|
address_pattern="{template_slug}.{unit_slug}.{function_slug}",
|
|
classification="internal",
|
|
allow_vacant_delivery=True,
|
|
actor_id="account-1",
|
|
)
|
|
session.commit()
|
|
|
|
listing = self.client.get("/api/v1/postbox/admin/templates")
|
|
self.assertEqual(200, listing.status_code, listing.text)
|
|
revision = listing.json()["templates"][0]["revisions"][0]
|
|
self.assertIsNone(revision["scope_structure_id"])
|
|
|
|
rejected = self.client.post(
|
|
"/api/v1/postbox/admin/templates",
|
|
json={
|
|
"slug": "new-subtree",
|
|
"name": "New subtree",
|
|
"scope_kind": "subtree",
|
|
"scope_id": "unit-1",
|
|
},
|
|
)
|
|
self.assertEqual(422, rejected.status_code, rejected.text)
|
|
|
|
def test_directory_delivery_message_and_receipt_round_trip(self) -> None:
|
|
directory = self.client.get("/api/v1/postbox/directory")
|
|
self.assertEqual(200, directory.status_code, directory.text)
|
|
self.assertEqual(self.postbox_id, directory.json()["postboxes"][0]["id"])
|
|
|
|
delivery = self.client.post(
|
|
"/api/v1/postbox/deliveries",
|
|
json={
|
|
"target": {"postbox_id": self.postbox_id},
|
|
"producer_module": "campaigns",
|
|
"producer_resource_type": "campaign_recipient",
|
|
"producer_resource_id": "recipient-1",
|
|
"idempotency_key": "campaign-1:recipient-1",
|
|
"subject": "Decision",
|
|
"body_text": "The decision is ready.",
|
|
"action_required": True,
|
|
},
|
|
)
|
|
self.assertEqual(201, delivery.status_code, delivery.text)
|
|
message_id = delivery.json()["message_id"]
|
|
|
|
messages = self.client.get(
|
|
"/api/v1/postbox/messages",
|
|
params={"postbox_id": self.postbox_id},
|
|
)
|
|
self.assertEqual(200, messages.status_code, messages.text)
|
|
self.assertEqual(1, messages.json()["total"])
|
|
self.assertEqual(message_id, messages.json()["messages"][0]["id"])
|
|
self.assertTrue(messages.json()["messages"][0]["metadata"]["action_required"])
|
|
|
|
filtered = self.client.get(
|
|
"/api/v1/postbox/messages",
|
|
params={
|
|
"postbox_id": self.postbox_id,
|
|
"q": "decision",
|
|
"state": "unread",
|
|
},
|
|
)
|
|
self.assertEqual(200, filtered.status_code, filtered.text)
|
|
self.assertEqual(1, filtered.json()["total"])
|
|
|
|
grouping = self.client.post(
|
|
"/api/v1/postbox/groupings",
|
|
json={
|
|
"name": "Assigned work",
|
|
"is_default": True,
|
|
"postbox_ids": [self.postbox_id],
|
|
},
|
|
)
|
|
self.assertEqual(201, grouping.status_code, grouping.text)
|
|
grouped_before_read = self.client.get("/api/v1/postbox/groupings")
|
|
self.assertEqual(200, grouped_before_read.status_code)
|
|
self.assertEqual(1, grouped_before_read.json()["groupings"][0]["total_count"])
|
|
self.assertEqual(1, grouped_before_read.json()["groupings"][0]["unread_count"])
|
|
|
|
acknowledged = self.client.patch(
|
|
f"/api/v1/postbox/messages/{message_id}/state",
|
|
json={"state": "acknowledged"},
|
|
)
|
|
self.assertEqual(200, acknowledged.status_code, acknowledged.text)
|
|
self.assertIsNotNone(acknowledged.json()["read_at"])
|
|
self.assertIsNotNone(acknowledged.json()["acknowledged_at"])
|
|
|
|
unread = self.client.get(
|
|
"/api/v1/postbox/messages",
|
|
params={
|
|
"postbox_id": self.postbox_id,
|
|
"state": "unread",
|
|
},
|
|
)
|
|
self.assertEqual(200, unread.status_code, unread.text)
|
|
self.assertEqual(0, unread.json()["total"])
|
|
grouped_after_read = self.client.get("/api/v1/postbox/groupings")
|
|
self.assertEqual(0, grouped_after_read.json()["groupings"][0]["unread_count"])
|
|
|
|
def test_routing_dry_run_explains_default_disabled_state(self) -> None:
|
|
response = self.client.post(
|
|
"/api/v1/postbox/routing/dry-run",
|
|
json={
|
|
"target": {"postbox_id": self.postbox_id},
|
|
"producer_module": "campaigns",
|
|
"classification": "internal",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertEqual("disabled", response.json()["status"])
|
|
self.assertEqual(
|
|
["hierarchy_routing_disabled"],
|
|
response.json()["diagnostics"],
|
|
)
|
|
|
|
def test_message_authoring_and_reply_are_idempotent_and_linked(self) -> None:
|
|
authored_payload = {
|
|
"postbox_id": self.postbox_id,
|
|
"idempotency_key": "compose-1",
|
|
"subject": "Status request",
|
|
"body_text": "Please provide a status update.",
|
|
"participants": [
|
|
{
|
|
"kind": "to",
|
|
"reference_type": "address",
|
|
"address": "team@example.invalid",
|
|
}
|
|
],
|
|
}
|
|
authored = self.client.post(
|
|
"/api/v1/postbox/messages",
|
|
json=authored_payload,
|
|
)
|
|
duplicate = self.client.post(
|
|
"/api/v1/postbox/messages",
|
|
json=authored_payload,
|
|
)
|
|
conflict = self.client.post(
|
|
"/api/v1/postbox/messages",
|
|
json={**authored_payload, "subject": "Different request"},
|
|
)
|
|
|
|
self.assertEqual(201, authored.status_code, authored.text)
|
|
self.assertEqual(201, duplicate.status_code, duplicate.text)
|
|
self.assertEqual(authored.json()["id"], duplicate.json()["id"])
|
|
self.assertEqual(409, conflict.status_code, conflict.text)
|
|
self.assertEqual("author", authored.json()["participants"][0]["kind"])
|
|
|
|
reply_payload = {
|
|
"idempotency_key": "reply-1",
|
|
"subject": "Re: Status request",
|
|
"body_text": "The work is complete.",
|
|
}
|
|
reply = self.client.post(
|
|
f"/api/v1/postbox/messages/{authored.json()['id']}/replies",
|
|
json=reply_payload,
|
|
)
|
|
duplicate_reply = self.client.post(
|
|
f"/api/v1/postbox/messages/{authored.json()['id']}/replies",
|
|
json=reply_payload,
|
|
)
|
|
|
|
self.assertEqual(201, reply.status_code, reply.text)
|
|
self.assertEqual(reply.json()["id"], duplicate_reply.json()["id"])
|
|
self.assertEqual(
|
|
authored.json()["id"],
|
|
reply.json()["in_reply_to_message_id"],
|
|
)
|
|
|
|
def test_mutable_admin_resources_require_strong_preconditions(self) -> None:
|
|
grouping = self.client.post(
|
|
"/api/v1/postbox/groupings",
|
|
json={
|
|
"name": "Work",
|
|
"is_default": True,
|
|
"postbox_ids": [self.postbox_id],
|
|
},
|
|
)
|
|
self.assertEqual(201, grouping.status_code, grouping.text)
|
|
grouping_data = grouping.json()
|
|
update_payload = {
|
|
"name": "Current work",
|
|
"is_default": True,
|
|
"postbox_ids": [self.postbox_id],
|
|
"base_revision": grouping_data["resource_revision"],
|
|
}
|
|
updated = self.client.put(
|
|
f"/api/v1/postbox/groupings/{grouping_data['id']}",
|
|
json=update_payload,
|
|
headers={"If-Match": grouping_data["etag"]},
|
|
)
|
|
stale = self.client.put(
|
|
f"/api/v1/postbox/groupings/{grouping_data['id']}",
|
|
json=update_payload,
|
|
headers={"If-Match": grouping_data["etag"]},
|
|
)
|
|
self.assertEqual(200, updated.status_code, updated.text)
|
|
self.assertEqual(2, updated.json()["resource_revision"])
|
|
self.assertEqual(412, stale.status_code, stale.text)
|
|
|
|
template = self.client.post(
|
|
"/api/v1/postbox/admin/templates",
|
|
json={
|
|
"slug": "case-intake",
|
|
"name": "Case intake",
|
|
"scope_kind": "tenant",
|
|
"name_pattern": "{unit_name} / {function_name}",
|
|
"address_pattern": "{template_slug}.{unit_slug}.{function_slug}",
|
|
"classification": "internal",
|
|
},
|
|
)
|
|
self.assertEqual(201, template.status_code, template.text)
|
|
template_data = template.json()
|
|
published = self.client.post(
|
|
f"/api/v1/postbox/admin/templates/{template_data['id']}/publish",
|
|
json={"base_revision": template_data["resource_revision"]},
|
|
headers={"If-Match": template_data["etag"]},
|
|
)
|
|
stale_retire = self.client.post(
|
|
f"/api/v1/postbox/admin/templates/{template_data['id']}/retire",
|
|
json={"base_revision": template_data["resource_revision"]},
|
|
headers={"If-Match": template_data["etag"]},
|
|
)
|
|
self.assertEqual(200, published.status_code, published.text)
|
|
self.assertEqual(412, stale_retire.status_code, stale_retire.text)
|
|
|
|
directory = self.client.get("/api/v1/postbox/admin/postboxes").json()
|
|
postbox = directory["postboxes"][0]
|
|
missing = self.client.request(
|
|
"DELETE",
|
|
f"/api/v1/postbox/admin/postboxes/{self.postbox_id}",
|
|
json={"base_revision": postbox["resource_revision"]},
|
|
)
|
|
archived = self.client.request(
|
|
"DELETE",
|
|
f"/api/v1/postbox/admin/postboxes/{self.postbox_id}",
|
|
json={"base_revision": postbox["resource_revision"]},
|
|
headers={"If-Match": postbox["etag"]},
|
|
)
|
|
self.assertEqual(428, missing.status_code, missing.text)
|
|
self.assertEqual(200, archived.status_code, archived.text)
|
|
self.assertEqual(2, archived.json()["resource_revision"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|