feat: implement governed service directory
This commit is contained in:
@@ -0,0 +1,391 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import unittest
|
||||
|
||||
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_portal.backend.manifest import get_manifest
|
||||
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_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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user