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

Add scoped postbox template previews

Browse Source
This commit is contained in:
zemion
2026-08-06 12:42:20 +02:00
parent c53646a8aa
commit d107c09f74
15 changed files with 1300 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_migration.py
+20 -7
View File
@@ -30,6 +30,10 @@ class PostboxMigrationTests(unittest.TestCase):
"govoplan_postbox.backend.migrations.versions."
"d8e3f6a9b2c5_postbox_content_protection"
)
scope_migration = importlib.import_module(
"govoplan_postbox.backend.migrations.versions."
"e9f4a7b2c5d8_v014_template_scope_preview"
)
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
@@ -39,17 +43,20 @@ class PostboxMigrationTests(unittest.TestCase):
occ_original = occ_migration.op
envelope_original = envelope_migration.op
protection_original = protection_migration.op
scope_original = scope_migration.op
migration.op = operations
route_migration.op = operations
occ_migration.op = operations
envelope_migration.op = operations
protection_migration.op = operations
scope_migration.op = operations
try:
migration.upgrade()
route_migration.upgrade()
occ_migration.upgrade()
envelope_migration.upgrade()
protection_migration.upgrade()
scope_migration.upgrade()
tables = set(inspect(connection).get_table_names())
self.assertIn("postboxes", tables)
self.assertIn("postbox_messages", tables)
@@ -75,14 +82,18 @@ class PostboxMigrationTests(unittest.TestCase):
"encryption_resource_id",
}.issubset(message_columns)
)
self.assertIn(
"encryption_vault_id",
template_revision_columns = {
column["name"]
for column in inspect(connection).get_columns(
"postbox_template_revisions"
)
}
self.assertTrue(
{
column["name"]
for column in inspect(connection).get_columns(
"postbox_template_revisions"
)
},
"encryption_vault_id",
"scope_structure_id",
"scope_relation_type_ids",
}.issubset(template_revision_columns)
)
self.assertIn("authoring_key", message_columns)
for table_name in (
@@ -110,6 +121,7 @@ class PostboxMigrationTests(unittest.TestCase):
route_columns
)
)
scope_migration.downgrade()
protection_migration.downgrade()
envelope_migration.downgrade()
occ_migration.downgrade()
@@ -128,6 +140,7 @@ class PostboxMigrationTests(unittest.TestCase):
occ_migration.op = occ_original
envelope_migration.op = envelope_original
protection_migration.op = protection_original
scope_migration.op = scope_original
finally:
engine.dispose()
tests/test_router.py
+73
View File
@@ -250,6 +250,63 @@ class PostboxRouterTests(unittest.TestCase):
self.assertEqual(response.status_code, 403)
self.assertIn("not active for this principal", response.text)
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)
@@ -289,6 +346,20 @@ class PostboxRouterTests(unittest.TestCase):
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"},
@@ -306,6 +377,8 @@ class PostboxRouterTests(unittest.TestCase):
)
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(
tests/test_service.py
+127
View File
@@ -899,6 +899,133 @@ class PostboxServiceTests(unittest.TestCase):
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,
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,
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,
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: