291 lines
10 KiB
Python
291 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
|
|
from govoplan_core.core.institutional import (
|
|
CAPABILITY_PARTY_RESOLVER,
|
|
EvidenceReference,
|
|
InstitutionalContextError,
|
|
InstitutionalReference,
|
|
LegalBasisReference,
|
|
PartyRepresentation,
|
|
PartySubjectReference,
|
|
ProcedureParty,
|
|
ServiceBinding,
|
|
ServiceDefinition,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_cases.backend.manifest import get_manifest
|
|
from govoplan_cases.backend.party_context import (
|
|
CasePartyCompatibilityRecord,
|
|
CasePartyContext,
|
|
)
|
|
from govoplan_cases.backend.service_intake import CaseServiceIntake
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
def reference(kind: str, object_id: str) -> InstitutionalReference:
|
|
owners = {
|
|
"organization_unit": "organizations",
|
|
"function": "organizations",
|
|
"mandate": "committee",
|
|
"jurisdiction": "organizations",
|
|
"service": "portal",
|
|
"case": "cases",
|
|
"party": "cases",
|
|
}
|
|
return InstitutionalReference(
|
|
kind=kind, # type: ignore[arg-type]
|
|
owner_module=owners[kind],
|
|
object_id=object_id,
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
valid_at=NOW,
|
|
)
|
|
|
|
|
|
def evidence() -> EvidenceReference:
|
|
return EvidenceReference(
|
|
kind="snapshot",
|
|
owner_module="addresses",
|
|
evidence_id="contact-snapshot-1",
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
captured_at=NOW,
|
|
)
|
|
|
|
|
|
def service() -> ServiceDefinition:
|
|
return ServiceDefinition(
|
|
reference=replace(reference("service", "permit"), version="4"),
|
|
key="permit.apply",
|
|
temporal=TemporalRevision(
|
|
revision="4",
|
|
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",),
|
|
legal_bases=(
|
|
LegalBasisReference(
|
|
kind="law",
|
|
authority="Example legislature",
|
|
reference="permit-law:3",
|
|
version="2026-01",
|
|
),
|
|
),
|
|
required_evidence_types=("identity", "application"),
|
|
deadline_refs=("deadline:permit:90d",),
|
|
responsible_organization_ref=reference("organization_unit", "unit-1"),
|
|
responsible_function_ref=reference("function", "function-1"),
|
|
mandate_ref=reference("mandate", "mandate-1"),
|
|
jurisdiction_refs=(reference("jurisdiction", "city-1"),),
|
|
bindings=(
|
|
ServiceBinding("case", "permit-application"),
|
|
ServiceBinding("form", "forms:permit-application"),
|
|
ServiceBinding("workflow", "workflow:permit-review"),
|
|
ServiceBinding("result", "decision:permit"),
|
|
),
|
|
publication_state="published",
|
|
)
|
|
|
|
|
|
def procedure_party(*, revoked_at: datetime | None = None) -> ProcedureParty:
|
|
represented = reference("party", "applicant")
|
|
representative = reference("party", "representative")
|
|
return ProcedureParty(
|
|
reference=representative,
|
|
procedure_ref=reference("case", "case-1"),
|
|
role="representative",
|
|
subject=PartySubjectReference(
|
|
kind="identity",
|
|
provider="identity",
|
|
subject_id="identity-2",
|
|
tenant_id="tenant-1",
|
|
version="3",
|
|
),
|
|
temporal=TemporalRevision(
|
|
revision="2",
|
|
valid_from=NOW - timedelta(days=2),
|
|
recorded_at=NOW - timedelta(days=2),
|
|
),
|
|
preferred_channels=("postbox",),
|
|
permitted_channels=("postbox", "mail"),
|
|
delivery_recipient=True,
|
|
representations=(
|
|
PartyRepresentation(
|
|
representative_party_ref=representative,
|
|
represented_party_ref=represented,
|
|
power_ref="power-1",
|
|
permitted_actions=("submit", "receive"),
|
|
temporal=TemporalRevision(
|
|
revision="1",
|
|
valid_from=NOW - timedelta(days=1),
|
|
recorded_at=NOW - timedelta(days=1),
|
|
),
|
|
evidence=(evidence(),),
|
|
revoked_at=revoked_at,
|
|
),
|
|
),
|
|
contact_snapshot_refs=("addresses:snapshot-1",),
|
|
evidence=(evidence(),),
|
|
)
|
|
|
|
|
|
class PartyProvider:
|
|
def __init__(self, party: ProcedureParty) -> None:
|
|
self.party = party
|
|
|
|
def list_procedure_parties(self, session, principal, *, procedure_ref, effective_at=None):
|
|
return (self.party,)
|
|
|
|
|
|
class DenyingPartyProvider:
|
|
def list_procedure_parties(self, session, principal, *, procedure_ref, effective_at=None):
|
|
raise InstitutionalContextError("Party provider access denied.")
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, provider: object | None = None) -> None:
|
|
self.provider = provider
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name == CAPABILITY_PARTY_RESOLVER and self.provider is not None
|
|
|
|
def capability(self, name: str) -> object:
|
|
return self.provider
|
|
|
|
|
|
class InstitutionalConsumerTests(unittest.TestCase):
|
|
def test_service_definition_drives_versioned_case_intake(self) -> None:
|
|
plan = CaseServiceIntake().plan(service(), case_id="case-1", effective_at=NOW)
|
|
|
|
self.assertEqual("permit-application", plan.case_type_ref)
|
|
self.assertEqual("4", plan.service_ref.version)
|
|
self.assertEqual("permit", plan.context.service_ref.object_id)
|
|
self.assertEqual("mandate-1", plan.context.mandate_ref.object_id)
|
|
self.assertEqual("city-1", plan.context.jurisdiction_refs[0].object_id)
|
|
self.assertEqual(("workflow:permit-review",), plan.workflow_refs)
|
|
|
|
def test_case_party_provider_drives_frozen_delivery_authority(self) -> None:
|
|
context = CasePartyContext(Registry(PartyProvider(procedure_party())))
|
|
party_set = context.resolve(
|
|
None,
|
|
None,
|
|
case_ref=reference("case", "case-1"),
|
|
effective_at=NOW,
|
|
)
|
|
targets = context.delivery_targets(party_set, channel="postbox")
|
|
|
|
self.assertEqual("provider", party_set.source)
|
|
self.assertEqual(("addresses:snapshot-1",), targets[0].contact_snapshot_refs)
|
|
self.assertEqual("applicant", targets[0].represented_party_refs[0].object_id)
|
|
|
|
def test_revoked_representation_is_not_delivery_authority(self) -> None:
|
|
context = CasePartyContext(
|
|
Registry(PartyProvider(procedure_party(revoked_at=NOW)))
|
|
)
|
|
party_set = context.resolve(
|
|
None,
|
|
None,
|
|
case_ref=reference("case", "case-1"),
|
|
effective_at=NOW,
|
|
)
|
|
|
|
self.assertEqual(
|
|
(),
|
|
context.delivery_targets(party_set, channel="postbox")[0].represented_party_refs,
|
|
)
|
|
|
|
def test_cases_only_compatibility_path_is_bounded(self) -> None:
|
|
subject = PartySubjectReference(
|
|
kind="organization",
|
|
provider="organizations",
|
|
subject_id="organization-1",
|
|
tenant_id="tenant-1",
|
|
)
|
|
party_set = CasePartyContext().resolve(
|
|
None,
|
|
None,
|
|
case_ref=reference("case", "case-1"),
|
|
effective_at=NOW,
|
|
compatibility=(
|
|
CasePartyCompatibilityRecord(
|
|
party_id="party-1",
|
|
role="applicant",
|
|
subject=subject,
|
|
valid_from=NOW - timedelta(days=1),
|
|
permitted_channels=("mail",),
|
|
delivery_recipient=True,
|
|
contact_snapshot_refs=("addresses:snapshot-2",),
|
|
),
|
|
),
|
|
)
|
|
|
|
self.assertEqual("compatibility", party_set.source)
|
|
self.assertEqual("organization-1", party_set.parties[0].subject.subject_id)
|
|
self.assertEqual((), party_set.parties[0].representations)
|
|
|
|
def test_cross_tenant_compatibility_subject_fails_closed(self) -> None:
|
|
with self.assertRaisesRegex(InstitutionalContextError, "another tenant"):
|
|
CasePartyContext().resolve(
|
|
None,
|
|
None,
|
|
case_ref=reference("case", "case-1"),
|
|
effective_at=NOW,
|
|
compatibility=(
|
|
CasePartyCompatibilityRecord(
|
|
party_id="party-1",
|
|
role="applicant",
|
|
subject=PartySubjectReference(
|
|
kind="identity",
|
|
provider="identity",
|
|
subject_id="identity-9",
|
|
tenant_id="tenant-2",
|
|
),
|
|
valid_from=NOW,
|
|
),
|
|
),
|
|
)
|
|
|
|
def test_provider_access_denial_never_falls_back_to_local_rows(self) -> None:
|
|
compatibility = CasePartyCompatibilityRecord(
|
|
party_id="party-1",
|
|
role="applicant",
|
|
subject=PartySubjectReference(
|
|
kind="identity",
|
|
provider="identity",
|
|
subject_id="identity-1",
|
|
tenant_id="tenant-1",
|
|
),
|
|
valid_from=NOW,
|
|
)
|
|
|
|
with self.assertRaisesRegex(InstitutionalContextError, "access denied"):
|
|
CasePartyContext(Registry(DenyingPartyProvider())).resolve(
|
|
None,
|
|
None,
|
|
case_ref=reference("case", "case-1"),
|
|
effective_at=NOW,
|
|
compatibility=(compatibility,),
|
|
)
|
|
|
|
def test_manifest_exposes_optional_party_provider(self) -> None:
|
|
manifest = get_manifest()
|
|
|
|
self.assertEqual("cases", manifest.id)
|
|
self.assertEqual((), manifest.dependencies)
|
|
self.assertIn(CAPABILITY_PARTY_RESOLVER, manifest.optional_capabilities)
|
|
self.assertIsNotNone(manifest.route_factory)
|
|
self.assertIsNotNone(manifest.migration_spec)
|
|
self.assertIn("cases.registry", manifest.capability_factories)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|