592 lines
21 KiB
Python
592 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.institutional import (
|
|
ActorRepresentationReference,
|
|
DecisionEffectReference,
|
|
EvidenceReference,
|
|
InformationGovernanceReference,
|
|
InstitutionalReference,
|
|
LegalBasisReference,
|
|
MandateDefinition,
|
|
PartyRepresentation,
|
|
PartySubjectReference,
|
|
ProcedureParty,
|
|
ServiceBinding,
|
|
ServiceDefinition,
|
|
TemporalRevision,
|
|
service_launch_capability,
|
|
)
|
|
from govoplan_cases.backend.party_context import CasePartyContext
|
|
from govoplan_cases.backend.db.models import (
|
|
CaseAccessGrant,
|
|
CaseIdentity,
|
|
CaseRecordRevision,
|
|
CaseStatusDefinition,
|
|
CaseTimelineEntry,
|
|
CaseTypeDefinition,
|
|
)
|
|
from govoplan_cases.backend.service import (
|
|
create_case_from_intake,
|
|
get_case,
|
|
upsert_case_status,
|
|
upsert_case_type,
|
|
)
|
|
from govoplan_cases.backend.service_intake import CaseServiceIntake
|
|
from govoplan_committee.backend.db.models import (
|
|
CommitteeDecisionProjection,
|
|
CommitteeWorkspaceEvent,
|
|
CommitteeWorkspaceRevision,
|
|
)
|
|
from govoplan_committee.backend.decision_path import (
|
|
CommitteeDecisionPath,
|
|
CommitteeDecisionProposal,
|
|
)
|
|
from govoplan_committee.backend.workspace import (
|
|
CommitteeWorkspaceRecord,
|
|
SqlCommitteeWorkspace,
|
|
get_workspace_object,
|
|
record_workspace_object,
|
|
)
|
|
from govoplan_portal.backend.service_directory import PortalServiceDirectory
|
|
from govoplan_decisions.backend.db.models import FormalDecisionRevision
|
|
from govoplan_decisions.backend.service import SqlDecisionRegistry
|
|
from govoplan_mandates.backend.db.models import MandateRevision
|
|
from govoplan_mandates.backend.service import SqlMandateResolver, record_mandate
|
|
from govoplan_parties.backend.db.models import ProcedurePartyRevision
|
|
from govoplan_parties.backend.service import SqlPartyResolver, record_procedure_party
|
|
from govoplan_services.backend.db.models import ServiceDefinitionRevision
|
|
from govoplan_services.backend.service import SqlServiceDefinitionProvider, record_service_definition
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
def _reference(
|
|
kind: str,
|
|
object_id: str,
|
|
*,
|
|
owner: str,
|
|
version: str = "1",
|
|
) -> InstitutionalReference:
|
|
return InstitutionalReference(
|
|
kind=kind, # type: ignore[arg-type]
|
|
owner_module=owner,
|
|
object_id=object_id,
|
|
tenant_id="tenant-1",
|
|
version=version,
|
|
valid_at=NOW,
|
|
)
|
|
|
|
|
|
def _evidence(kind: str, evidence_id: str, owner: str) -> EvidenceReference:
|
|
return EvidenceReference(
|
|
kind=kind, # type: ignore[arg-type]
|
|
owner_module=owner,
|
|
evidence_id=evidence_id,
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
checksum=f"sha256:{evidence_id}",
|
|
captured_at=NOW,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, capabilities: dict[str, object]) -> None:
|
|
self.capabilities = capabilities
|
|
|
|
def has(self, module_id: str) -> bool:
|
|
return module_id in {
|
|
"portal",
|
|
"cases",
|
|
"committee",
|
|
"postbox",
|
|
"services",
|
|
"parties",
|
|
"mandates",
|
|
"decisions",
|
|
}
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name in self.capabilities
|
|
|
|
def capability(self, name: str) -> object:
|
|
return self.capabilities[name]
|
|
|
|
|
|
class InstitutionalGovernanceJourneyTests(unittest.TestCase):
|
|
def test_service_to_formal_outcome_retains_governed_context(self) -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
for table in (
|
|
ServiceDefinitionRevision.__table__,
|
|
CaseStatusDefinition.__table__,
|
|
CaseTypeDefinition.__table__,
|
|
CaseIdentity.__table__,
|
|
CaseRecordRevision.__table__,
|
|
CaseAccessGrant.__table__,
|
|
CaseTimelineEntry.__table__,
|
|
ProcedurePartyRevision.__table__,
|
|
MandateRevision.__table__,
|
|
FormalDecisionRevision.__table__,
|
|
CommitteeWorkspaceRevision.__table__,
|
|
CommitteeWorkspaceEvent.__table__,
|
|
CommitteeDecisionProjection.__table__,
|
|
):
|
|
table.create(engine)
|
|
session = Session(engine)
|
|
principal = _Principal()
|
|
self.addCleanup(engine.dispose)
|
|
self.addCleanup(session.close)
|
|
|
|
organization = _reference(
|
|
"organization_unit",
|
|
"permit-office",
|
|
owner="organizations",
|
|
)
|
|
function = _reference("function", "permit-officer", owner="organizations")
|
|
jurisdiction = _reference(
|
|
"jurisdiction",
|
|
"city-1",
|
|
owner="organizations",
|
|
)
|
|
mandate_ref = _reference(
|
|
"mandate",
|
|
"permit-mandate",
|
|
owner="mandates",
|
|
version="4",
|
|
)
|
|
legal_basis = LegalBasisReference(
|
|
kind="law",
|
|
authority="Example legislature",
|
|
reference="permit-law:3",
|
|
version="2026-01",
|
|
effective_from=NOW - timedelta(days=100),
|
|
)
|
|
service = ServiceDefinition(
|
|
reference=_reference(
|
|
"service",
|
|
"permit-service",
|
|
owner="services",
|
|
version="5",
|
|
),
|
|
key="permit.apply",
|
|
temporal=TemporalRevision(
|
|
revision="5",
|
|
valid_from=NOW - timedelta(days=1),
|
|
valid_to=NOW + timedelta(days=30),
|
|
recorded_at=NOW - timedelta(days=2),
|
|
change_reason="Service published.",
|
|
),
|
|
title="Apply for a permit",
|
|
audience=("resident",),
|
|
legal_bases=(legal_basis,),
|
|
required_evidence_types=("application", "identity"),
|
|
channels=("portal", "postbox"),
|
|
responsible_organization_ref=organization,
|
|
responsible_function_ref=function,
|
|
mandate_ref=mandate_ref,
|
|
jurisdiction_refs=(jurisdiction,),
|
|
bindings=(
|
|
ServiceBinding("case", "permit-application"),
|
|
ServiceBinding("workflow", "workflow:permit-review"),
|
|
ServiceBinding("result", "decision:permit"),
|
|
),
|
|
remedy_refs=("review:administrative-court",),
|
|
publication_state="published",
|
|
)
|
|
record_service_definition(session, principal, definition=service)
|
|
service_registry = _Registry(
|
|
{
|
|
"services.definitions": SqlServiceDefinitionProvider(),
|
|
service_launch_capability("case"): object(),
|
|
}
|
|
)
|
|
entry = PortalServiceDirectory(service_registry).list_entries(
|
|
session,
|
|
principal,
|
|
tenant_id="tenant-1",
|
|
effective_at=NOW,
|
|
audiences=("resident",),
|
|
)[0]
|
|
self.assertTrue(entry.available)
|
|
intake = CaseServiceIntake().plan(
|
|
entry.definition,
|
|
case_id="case-1",
|
|
effective_at=NOW,
|
|
)
|
|
application_evidence = _evidence("document", "application-1", "files")
|
|
upsert_case_status(
|
|
session,
|
|
principal,
|
|
status_key="intake",
|
|
label="Intake",
|
|
)
|
|
upsert_case_status(
|
|
session,
|
|
principal,
|
|
status_key="decided",
|
|
label="Decided",
|
|
category="decided",
|
|
)
|
|
upsert_case_type(
|
|
session,
|
|
principal,
|
|
type_key="permit-application",
|
|
label="Permit application",
|
|
initial_status_key="intake",
|
|
allowed_status_keys=("intake", "decided"),
|
|
)
|
|
case_record = create_case_from_intake(
|
|
session,
|
|
principal,
|
|
plan=intake,
|
|
case_number="PERMIT-2026-0001",
|
|
title="Permit application",
|
|
status_key=None,
|
|
opened_at=NOW,
|
|
recorded_at=NOW,
|
|
change_reason="Portal application received.",
|
|
idempotency_key="journey-case-create",
|
|
evidence_refs=(application_evidence,),
|
|
deadline_at=NOW + timedelta(days=30),
|
|
)
|
|
|
|
applicant = _reference("party", "applicant", owner="parties")
|
|
representative = _reference("party", "representative", owner="parties")
|
|
address_evidence = _evidence(
|
|
"snapshot",
|
|
"address-snapshot-1",
|
|
"addresses",
|
|
)
|
|
representative_party = ProcedureParty(
|
|
reference=representative,
|
|
procedure_ref=case_record.reference,
|
|
role="representative",
|
|
subject=PartySubjectReference(
|
|
kind="identity",
|
|
provider="identity",
|
|
subject_id="identity-2",
|
|
tenant_id="tenant-1",
|
|
version="2",
|
|
),
|
|
temporal=TemporalRevision(
|
|
revision="1",
|
|
valid_from=NOW - timedelta(days=1),
|
|
recorded_at=NOW - timedelta(days=1),
|
|
change_reason="Representative added to the procedure.",
|
|
),
|
|
permitted_channels=("postbox",),
|
|
preferred_channels=("postbox",),
|
|
delivery_recipient=True,
|
|
representations=(
|
|
PartyRepresentation(
|
|
representative_party_ref=representative,
|
|
represented_party_ref=applicant,
|
|
power_ref="power-1",
|
|
permitted_actions=("submit", "receive"),
|
|
temporal=TemporalRevision(
|
|
revision="1",
|
|
valid_from=NOW - timedelta(days=1),
|
|
recorded_at=NOW - timedelta(days=1),
|
|
change_reason="Representation power recorded.",
|
|
),
|
|
evidence=(address_evidence,),
|
|
),
|
|
),
|
|
contact_snapshot_refs=("addresses:snapshot-1",),
|
|
evidence=(address_evidence,),
|
|
)
|
|
record_procedure_party(session, principal, party=representative_party)
|
|
party_context = CasePartyContext(
|
|
_Registry({"parties.resolver": SqlPartyResolver()})
|
|
)
|
|
parties = party_context.resolve(
|
|
session,
|
|
principal,
|
|
case_ref=case_record.reference,
|
|
effective_at=NOW,
|
|
)
|
|
delivery_target = party_context.delivery_targets(
|
|
parties,
|
|
channel="postbox",
|
|
)[0]
|
|
|
|
mandate = MandateDefinition(
|
|
reference=mandate_ref,
|
|
temporal=TemporalRevision(
|
|
revision="4",
|
|
valid_from=NOW - timedelta(days=30),
|
|
valid_to=NOW + timedelta(days=30),
|
|
recorded_at=NOW - timedelta(days=31),
|
|
change_reason="Permit authority delegated.",
|
|
),
|
|
task_types=("committee.formal_decision",),
|
|
authority_types=("permit",),
|
|
organization_unit_refs=(organization,),
|
|
function_refs=(function,),
|
|
jurisdiction_refs=(jurisdiction,),
|
|
legal_bases=(legal_basis,),
|
|
evidence=(_evidence("record", "mandate-record-4", "mandates"),),
|
|
authority_ceiling="permit:standard",
|
|
)
|
|
record_mandate(session, principal, definition=mandate)
|
|
workspace = SqlCommitteeWorkspace()
|
|
body = CommitteeWorkspaceRecord(
|
|
tenant_id="tenant-1",
|
|
object_kind="body",
|
|
object_id="permit-board",
|
|
revision=1,
|
|
state="active",
|
|
title="Permit board",
|
|
recorded_at=NOW,
|
|
change_reason="Permit board configured.",
|
|
attributes={
|
|
"organization_unit_ref": organization.to_dict(),
|
|
"function_refs": [function.to_dict()],
|
|
"quorum": {"minimum_count": 1},
|
|
},
|
|
)
|
|
meeting = CommitteeWorkspaceRecord(
|
|
tenant_id="tenant-1",
|
|
object_kind="meeting",
|
|
object_id="meeting-1",
|
|
revision=1,
|
|
state="open",
|
|
title="Permit board meeting",
|
|
parent_id=body.object_id,
|
|
recorded_at=NOW,
|
|
change_reason="Meeting opened.",
|
|
attributes={
|
|
"starts_at": NOW.isoformat(),
|
|
"ends_at": (NOW + timedelta(hours=1)).isoformat(),
|
|
},
|
|
context=case_record.context,
|
|
)
|
|
agenda = CommitteeWorkspaceRecord(
|
|
tenant_id="tenant-1",
|
|
object_kind="agenda_item",
|
|
object_id="item-1",
|
|
revision=1,
|
|
state="deliberating",
|
|
title="Permit application",
|
|
parent_id=meeting.object_id,
|
|
recorded_at=NOW,
|
|
change_reason="Agenda item entered deliberation.",
|
|
attributes={
|
|
"position": 1,
|
|
"subject_refs": [case_record.reference.to_dict()],
|
|
},
|
|
context=case_record.context,
|
|
evidence=(application_evidence,),
|
|
)
|
|
approval_ref = _reference("approval", "approval-1", owner="approvals")
|
|
vote = CommitteeWorkspaceRecord(
|
|
tenant_id="tenant-1",
|
|
object_kind="vote",
|
|
object_id="vote-1",
|
|
revision=1,
|
|
state="closed",
|
|
title="Vote on permit application",
|
|
parent_id=agenda.object_id,
|
|
recorded_at=NOW,
|
|
change_reason="Vote result accepted.",
|
|
attributes={
|
|
"method": "recorded",
|
|
"choices": ["yes", "no"],
|
|
"eligible_count": 3,
|
|
"cast_count": 3,
|
|
"counts": {"yes": 3, "no": 0},
|
|
"quorum_met": True,
|
|
"approval_ref": approval_ref.to_dict(),
|
|
},
|
|
context=case_record.context,
|
|
evidence=(_evidence("record", "vote-result-1", "committee"),),
|
|
)
|
|
for record, key in (
|
|
(body, "journey-body"),
|
|
(meeting, "journey-meeting"),
|
|
(agenda, "journey-agenda"),
|
|
(vote, "journey-vote"),
|
|
):
|
|
record_workspace_object(
|
|
session,
|
|
principal,
|
|
record=record,
|
|
idempotency_key=key,
|
|
)
|
|
decision_registry = SqlDecisionRegistry()
|
|
decision_result = CommitteeDecisionPath(
|
|
_Registry(
|
|
{
|
|
"mandates.resolver": SqlMandateResolver(),
|
|
"decisions.registry": decision_registry,
|
|
"committee.workspace": workspace,
|
|
}
|
|
)
|
|
).decide(
|
|
session,
|
|
principal,
|
|
proposal=CommitteeDecisionProposal(
|
|
tenant_id="tenant-1",
|
|
decision_id="decision-1",
|
|
revision="1",
|
|
effective_at=NOW,
|
|
meeting_ref="meeting-1",
|
|
agenda_item_ref="item-1",
|
|
decision_type="permit",
|
|
subject_refs=(case_record.reference,),
|
|
organization_unit_ref=organization,
|
|
function_ref=function,
|
|
actor=ActorRepresentationReference(
|
|
tenant_id="tenant-1",
|
|
account_id="account-1",
|
|
identity_id="identity-1",
|
|
represented_function_ref=function,
|
|
mandate_ref=mandate_ref,
|
|
),
|
|
approval_refs=(
|
|
approval_ref,
|
|
),
|
|
fact_evidence=(application_evidence, address_evidence),
|
|
legal_bases=(legal_basis,),
|
|
operative_result="Permit granted.",
|
|
reasoning="The application satisfies the effective rule.",
|
|
case_ref=case_record.reference,
|
|
jurisdiction_refs=(jurisdiction,),
|
|
party_refs=(applicant, representative),
|
|
record_refs=(
|
|
_reference("record", "record-1", owner="audit"),
|
|
),
|
|
remedy_refs=service.remedy_refs,
|
|
review_refs=("review:administrative-court",),
|
|
information_governance=InformationGovernanceReference(
|
|
classification="restricted",
|
|
purposes=("permit-decision", "party-delivery"),
|
|
legal_basis_refs=("permit-law:3@2026-01",),
|
|
retention_policy_ref="records:permit",
|
|
disclosure_state="partly_disclosable",
|
|
),
|
|
),
|
|
observed_effects=(
|
|
DecisionEffectReference(
|
|
effect_key="postbox.deliver_decision",
|
|
state="confirmed",
|
|
resource_refs=(
|
|
f"postbox:{delivery_target.party_ref.object_id}",
|
|
),
|
|
audit_event_refs=("audit:delivery-1",),
|
|
evidence_refs=(address_evidence.evidence_id,),
|
|
),
|
|
),
|
|
)
|
|
decided_agenda = replace(
|
|
agenda,
|
|
revision=2,
|
|
state="decided",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Formal Decision recorded.",
|
|
attributes={
|
|
**dict(agenda.attributes),
|
|
"decision_ref": decision_result.decision.reference.to_dict(),
|
|
},
|
|
)
|
|
record_workspace_object(
|
|
session,
|
|
principal,
|
|
record=decided_agenda,
|
|
expected_revision=1,
|
|
idempotency_key="journey-agenda-decided",
|
|
)
|
|
closed_meeting = replace(
|
|
meeting,
|
|
revision=2,
|
|
state="closed",
|
|
recorded_at=NOW + timedelta(hours=1),
|
|
change_reason="All agenda items completed.",
|
|
)
|
|
record_workspace_object(
|
|
session,
|
|
principal,
|
|
record=closed_meeting,
|
|
expected_revision=1,
|
|
idempotency_key="journey-meeting-closed",
|
|
)
|
|
minute = CommitteeWorkspaceRecord(
|
|
tenant_id="tenant-1",
|
|
object_kind="minute",
|
|
object_id="minute-1",
|
|
revision=1,
|
|
state="accepted",
|
|
title="Accepted permit board minutes",
|
|
parent_id=meeting.object_id,
|
|
recorded_at=NOW + timedelta(hours=2),
|
|
change_reason="Minutes approved.",
|
|
attributes={
|
|
"content_ref": _reference(
|
|
"record",
|
|
"meeting-minutes-1",
|
|
owner="records",
|
|
).to_dict(),
|
|
"approval_ref": _reference(
|
|
"approval",
|
|
"minutes-approval-1",
|
|
owner="approvals",
|
|
).to_dict(),
|
|
},
|
|
context=case_record.context,
|
|
evidence=(_evidence("record", "minutes-proof-1", "records"),),
|
|
)
|
|
record_workspace_object(
|
|
session,
|
|
principal,
|
|
record=minute,
|
|
idempotency_key="journey-minute",
|
|
)
|
|
reconstruction = decision_result.reconstruction_payload()["decision"]
|
|
persisted = decision_registry.get_decision(
|
|
session,
|
|
principal,
|
|
reference=decision_result.decision.reference,
|
|
)
|
|
|
|
self.assertTrue(decision_result.persisted_by_decision_registry)
|
|
self.assertEqual(decision_result.decision, persisted)
|
|
self.assertEqual(
|
|
case_record,
|
|
get_case(session, principal, case_id="case-1"),
|
|
)
|
|
self.assertEqual(
|
|
"decided",
|
|
get_workspace_object(
|
|
session,
|
|
principal,
|
|
object_kind="agenda_item",
|
|
object_id="item-1",
|
|
).state,
|
|
)
|
|
self.assertEqual(service.reference, intake.context.service_ref)
|
|
self.assertEqual("applicant", delivery_target.represented_party_refs[0].object_id)
|
|
self.assertEqual("permit-mandate", reconstruction["authority_context"]["mandate_ref"]["object_id"])
|
|
self.assertEqual("city-1", reconstruction["authority_context"]["jurisdiction_refs"][0]["object_id"])
|
|
self.assertEqual("confirmed", reconstruction["observed_effects"][0]["state"])
|
|
self.assertEqual("audit:delivery-1", reconstruction["observed_effects"][0]["audit_event_refs"][0])
|
|
self.assertEqual("application-1", reconstruction["fact_evidence"][0]["evidence_id"])
|
|
self.assertEqual("The application satisfies the effective rule.", reconstruction["reasoning"])
|
|
self.assertEqual("review:administrative-court", reconstruction["review_refs"][0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|