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, CAPABILITY_SERVICE_DEFINITIONS, EvidenceReference, InstitutionalContextError, InstitutionalReference, ServiceBinding, ServiceAvailabilityAssessment, ServiceAvailabilityRequirement, ServiceDefinition, ServiceLaunchResult, TemporalRevision, ) from govoplan_core.core.access import ( CAPABILITY_ACCESS_SEMANTIC_DIRECTORY, FunctionAssignmentRef, 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, ) NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC) def service( *, state: str = "published", requirements: tuple[ServiceAvailabilityRequirement, ...] = (), ) -> ServiceDefinition: return ServiceDefinition( reference=InstitutionalReference( kind="service", owner_module="portal", object_id="permit", tenant_id="tenant-1", version="3", ), key="permit.apply", temporal=TemporalRevision( revision="3", valid_from=NOW - timedelta(days=1), valid_to=NOW + timedelta(days=1), recorded_at=NOW - timedelta(days=2), ), title="Apply for a permit", audience=("resident",), bindings=( ServiceBinding("capability", "cases.service_intake"), ServiceBinding("case", "permit-application"), ), availability_requirements=requirements, publication_state=state, # type: ignore[arg-type] availability_explanation_ref="docs:permit-requirements", ) class Provider: def __init__(self, definition: ServiceDefinition | None = None) -> None: self.definition = definition or service() def get_service_definition(self, session, principal, *, reference, effective_at=None): return self.definition def list_service_definitions(self, session, principal, *, tenant_id, query="", limit=100): return (self.definition,) class Launcher: def launch_service(self, session, principal, *, definition, request): return ServiceLaunchResult( service_ref=definition.reference, binding=request.binding, state="started", target_ref=InstitutionalReference( kind="case", owner_module="cases", object_id="case-1", tenant_id="tenant-1", version="1", ), href="/cases/case-1", metadata={}, ) class Registry: def __init__( self, *, intake: bool = False, evaluator: object | None = None, definition: ServiceDefinition | None = None, ) -> None: self.capabilities = { CAPABILITY_SERVICE_DEFINITIONS: Provider(definition) } if intake: self.capabilities["cases.service_intake"] = object() self.capabilities["cases.service_launcher"] = Launcher() if evaluator is not None: self.capabilities[CAPABILITY_SERVICE_AVAILABILITY] = evaluator def has(self, module_id: str) -> bool: return False def has_capability(self, name: str) -> bool: return name in self.capabilities def capability(self, name: str) -> object: return self.capabilities[name] class SemanticDirectory: def get_account(self, account_id): return None def get_user(self, user_id): return None def get_users(self, user_ids): return {} def users_for_tenant(self, tenant_id): return () def get_group(self, group_id): return None def get_groups(self, group_ids): return {} def groups_for_tenant(self, tenant_id): return () def groups_for_user(self, user_id, *, tenant_id): return () def display_label(self, subject): return None def get_identity(self, identity_id): return None def accounts_for_identity(self, identity_id): return () def get_organization_unit(self, organization_unit_id): return None def organization_units_for_tenant(self, tenant_id): return () def get_function(self, function_id): return FunctionRef( id=function_id, tenant_id="tenant-1", organization_unit_id="unit-1", slug="permit-clerk", name="Permit clerk", ) def functions_for_organization_unit( self, organization_unit_id, *, include_subunits=False ): return () def get_function_assignment(self, assignment_id): return FunctionAssignmentRef( id=assignment_id, tenant_id="tenant-1", account_id="account-1", function_id="function-1", organization_unit_id="unit-1", ) def function_assignments_for_account(self, account_id, *, tenant_id=None): return () 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, None, tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), ) self.assertEqual(1, len(entries)) self.assertTrue(entries[0].available) self.assertEqual("case", entries[0].entry_binding.kind) def test_missing_capability_is_explained_and_filterable(self) -> None: directory = PortalServiceDirectory(Registry()) entry = directory.project( service(), tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), ) self.assertEqual("unavailable", entry.state) self.assertIn( "service.required_capability.missing:cases.service_intake", entry.reason_codes, ) self.assertEqual( (), directory.list_entries( None, None, tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), include_unavailable=False, ), ) def test_exact_available_service_launch_uses_owner_launcher(self) -> None: definition = replace(service(), audience=("public",)) result = PortalServiceDirectory( Registry(intake=True, definition=definition) ).launch_service( None, None, reference=definition.reference, requested_at=NOW, idempotency_key="launch-1", parameters={}, ) self.assertEqual("started", result.state) self.assertEqual("case-1", result.target_ref.object_id) def test_non_applicable_or_unpublished_service_is_hidden(self) -> None: directory = PortalServiceDirectory(Registry(intake=True)) self.assertEqual( "hidden", directory.project( service(state="draft"), tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), ).state, ) self.assertEqual( "hidden", directory.project( service(), tenant_id="tenant-1", effective_at=NOW, audiences=("operator",), ).state, ) def test_cross_tenant_projection_fails_closed(self) -> None: with self.assertRaisesRegex(InstitutionalContextError, "another tenant"): PortalServiceDirectory().project( service(), tenant_id="tenant-2", effective_at=NOW, ) def test_governed_requirements_explain_or_conceal_failures(self) -> None: definition = service( requirements=( ServiceAvailabilityRequirement( "policy", "permit-access", explanation_ref="docs:permit-policy", ), ServiceAvailabilityRequirement( "maintenance", "permit-backend", failure_state="hidden", ), ) ) assessment = ServiceAvailabilityAssessment( requirement_states={"policy:permit-access": False}, reason_codes=("policy.denied",), ) entry = PortalServiceDirectory(Registry(intake=True)).project( definition, tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), assessment=assessment, ) self.assertEqual("hidden", entry.state) self.assertIn( "service.requirement.failed:policy:permit-access", entry.reason_codes, ) self.assertIn( "service.requirement.unknown:maintenance:permit-backend", entry.reason_codes, ) self.assertIn("policy.denied", entry.reason_codes) def test_availability_evidence_cannot_cross_tenants(self) -> None: assessment = ServiceAvailabilityAssessment( requirement_states={}, evidence=( EvidenceReference( kind="record", owner_module="policy", evidence_id="decision-1", tenant_id="tenant-2", ), ), ) with self.assertRaisesRegex(InstitutionalContextError, "another tenant"): PortalServiceDirectory(Registry(intake=True)).project( service(), tenant_id="tenant-1", effective_at=NOW, audiences=("resident",), assessment=assessment, ) def test_principal_audiences_resolve_active_function_semantics(self) -> None: registry = Registry(intake=True) registry.capabilities[CAPABILITY_ACCESS_SEMANTIC_DIRECTORY] = ( SemanticDirectory() ) principal = PrincipalRef( account_id="account-1", membership_id="membership-1", tenant_id="tenant-1", identity_id="identity-1", group_ids=frozenset({"group-1"}), role_ids=frozenset({"role-1"}), function_assignment_ids=frozenset({"assignment-1"}), ) audiences = principal_audiences( registry, None, principal, tenant_id="tenant-1", effective_at=NOW, ) self.assertIn("authenticated", audiences) self.assertIn("function:permit-clerk", audiences) self.assertIn("organization-unit:unit-1", audiences) self.assertIn("role:role-1", audiences) def test_empty_audience_projection_fails_closed(self) -> None: entry = PortalServiceDirectory(Registry(intake=True)).project( service(), tenant_id="tenant-1", effective_at=NOW, ) self.assertEqual("hidden", entry.state) self.assertIn("service.audience.not_applicable", entry.reason_codes) def test_manifest_exposes_api_and_service_directory_webui(self) -> None: manifest = get_manifest() self.assertEqual("portal", manifest.id) self.assertIsNotNone(manifest.route_factory) self.assertIsNotNone(manifest.frontend) self.assertEqual("@govoplan/portal-webui", manifest.frontend.package_name) self.assertIn( "portal:service:read", {item.scope for item in manifest.permissions}, ) self.assertIn("portal.service_directory", manifest.capability_factories) self.assertIsNotNone(manifest.public_tenant_resolver) self.assertIn( "/portal/status/:trackingId", {route.path for route in manifest.frontend.public_routes}, ) self.assertIn( "application_status.projection", {item.name for item in manifest.requires_interfaces}, ) self.assertFalse( any( name.startswith("privacy.dsar.") for name in manifest.capability_factories ) ) dsar_topic = next( item for item in manifest.documentation if item.id == "portal.data-subject-requests" ) self.assertEqual( "not_applicable_no_persistence", dsar_topic.metadata["dsar_coverage"], ) self.assertTrue({"admin", "user"}.issubset(dsar_topic.documentation_types)) topics = {topic.id: topic for topic in manifest.documentation} self.assertTrue( all( all( topic.translations.get("de", {}).get(field) for field in ("title", "summary", "body") ) for topic in topics.values() ) ) workflow = topics["portal.service-directory"] self.assertEqual("workflow", workflow.metadata["kind"]) self.assertTrue(workflow.conditions) self.assertTrue( all( condition.required_scopes or condition.any_scopes for condition in workflow.conditions ) ) self.assertEqual("reference", dsar_topic.metadata["kind"]) if __name__ == "__main__": unittest.main()