feat: project role-bound postboxes

This commit is contained in:
2026-08-07 14:53:52 +02:00
parent 1bc367a454
commit fea8e4b5e6
7 changed files with 274 additions and 1 deletions
+46
View File
@@ -3,6 +3,8 @@ from __future__ import annotations
from dataclasses import replace
from datetime import UTC, datetime, timedelta
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from govoplan_core.core.institutional import (
CAPABILITY_SERVICE_AVAILABILITY,
@@ -23,7 +25,10 @@ from govoplan_core.core.access import (
FunctionRef,
PrincipalRef,
)
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.postbox import PostboxDirectoryEntryRef, PostboxPortalEntryRef
from govoplan_portal.backend.manifest import get_manifest
from govoplan_portal.backend.router import api_list_portal_postboxes
from govoplan_portal.backend.service_directory import (
PortalServiceDirectory,
principal_audiences,
@@ -189,6 +194,47 @@ class SemanticDirectory:
class PortalServiceDirectoryTests(unittest.TestCase):
def test_portal_postbox_endpoint_projects_optional_provider_entries(self) -> None:
principal = ApiPrincipal(
principal=PrincipalRef(
account_id="account-1",
membership_id="membership-1",
tenant_id="tenant-1",
scopes=frozenset({"portal:service:read"}),
),
account=SimpleNamespace(id="account-1"),
user=SimpleNamespace(id="membership-1"),
)
entry = PostboxPortalEntryRef(
postbox=PostboxDirectoryEntryRef(
id="postbox-1",
tenant_id="tenant-1",
address="clerk.district",
address_key="clerk.district",
name="District / Clerk",
status="active",
classification="internal",
),
unread_count=3,
)
provider = SimpleNamespace(
list_portal_entries=lambda *_args, **_kwargs: (entry,)
)
with patch(
"govoplan_portal.backend.router.postbox_portal_projection_provider",
return_value=provider,
):
response = api_list_portal_postboxes(
limit=100,
session=object(),
principal=principal,
)
self.assertTrue(response.provider_available)
self.assertEqual("postbox-1", response.postboxes[0].postbox["id"])
self.assertEqual(3, response.postboxes[0].unread_count)
def test_provider_definition_is_available_when_requirements_exist(self) -> None:
entries = PortalServiceDirectory(Registry(intake=True)).list_entries(
None,