govoplan-portal/tests/test_service_directory.py

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()