447 lines
14 KiB
Python
447 lines
14 KiB
Python
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},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|