Integrate formal decisions with cases
This commit is contained in:
@@ -0,0 +1,350 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import unittest
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.access import (
|
||||
CAPABILITY_ACCESS_SEMANTIC_DIRECTORY,
|
||||
FunctionAssignmentRef,
|
||||
)
|
||||
from govoplan_core.core.institutional import (
|
||||
CAPABILITY_DECISION_REGISTRY,
|
||||
CAPABILITY_MANDATE_RESOLVER,
|
||||
EvidenceReference,
|
||||
FormalDecision,
|
||||
GovernedContextEnvelope,
|
||||
InstitutionalReference,
|
||||
LegalBasisReference,
|
||||
MandateDefinition,
|
||||
MandateResolution,
|
||||
TemporalRevision,
|
||||
)
|
||||
from govoplan_cases.backend.db.models import (
|
||||
CaseAccessGrant,
|
||||
CaseIdentity,
|
||||
CaseRecordRevision,
|
||||
CaseStatusDefinition,
|
||||
CaseTimelineEntry,
|
||||
CaseTypeDefinition,
|
||||
)
|
||||
from govoplan_cases.backend.decision_path import (
|
||||
CaseDecisionCommand,
|
||||
CaseDecisionError,
|
||||
CaseDecisionPath,
|
||||
)
|
||||
from govoplan_cases.backend.domain import CaseRecord
|
||||
from govoplan_cases.backend.service import (
|
||||
create_case,
|
||||
get_case,
|
||||
upsert_case_status,
|
||||
upsert_case_type,
|
||||
)
|
||||
|
||||
|
||||
NOW = datetime(2026, 8, 6, 9, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Principal:
|
||||
tenant_id: str = "tenant-1"
|
||||
account_id: str = "account-1"
|
||||
identity_id: str = "identity-1"
|
||||
acting_assignment_id: str | None = "assignment-1"
|
||||
function_assignment_ids: tuple[str, ...] = ("assignment-1",)
|
||||
scopes: frozenset[str] = frozenset(
|
||||
{"cases:case:read", "cases:case:update", "decisions:decision:write"}
|
||||
)
|
||||
|
||||
|
||||
class Registry:
|
||||
def __init__(self, providers: dict[str, object]) -> None:
|
||||
self.providers = providers
|
||||
|
||||
def has_capability(self, name: str) -> bool:
|
||||
return name in self.providers
|
||||
|
||||
def capability(self, name: str) -> object:
|
||||
return self.providers[name]
|
||||
|
||||
|
||||
class AccessDirectory:
|
||||
def get_function_assignment(self, assignment_id: str):
|
||||
if assignment_id != "assignment-1":
|
||||
return None
|
||||
return FunctionAssignmentRef(
|
||||
id=assignment_id,
|
||||
tenant_id="tenant-1",
|
||||
account_id="account-1",
|
||||
identity_id="identity-1",
|
||||
function_id="permit-officer",
|
||||
organization_unit_id="permits",
|
||||
valid_from=NOW - timedelta(days=30),
|
||||
valid_until=NOW + timedelta(days=30),
|
||||
)
|
||||
|
||||
|
||||
class Mandates:
|
||||
def __init__(self, mandate: MandateDefinition) -> None:
|
||||
self.mandate = mandate
|
||||
|
||||
def resolve_mandate(self, session, principal, *, request):
|
||||
del session, principal
|
||||
return MandateResolution(competent=True, mandates=(self.mandate,))
|
||||
|
||||
|
||||
class Decisions:
|
||||
def __init__(self) -> None:
|
||||
self.items: dict[tuple[str, str], FormalDecision] = {}
|
||||
|
||||
def get_decision(self, session, principal, *, reference):
|
||||
del session, principal
|
||||
return self.items.get((reference.object_id, reference.version or ""))
|
||||
|
||||
def record_decision(self, session, principal, *, decision, expected_revision=None):
|
||||
del session, principal, expected_revision
|
||||
key = (decision.reference.object_id, decision.reference.version or "")
|
||||
existing = self.items.get(key)
|
||||
if existing is not None and existing != decision:
|
||||
raise ValueError("A different Decision payload already uses this revision.")
|
||||
self.items[key] = decision
|
||||
return decision
|
||||
|
||||
|
||||
def ref(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 legal_basis() -> LegalBasisReference:
|
||||
return LegalBasisReference(
|
||||
kind="law",
|
||||
authority="Example legislature",
|
||||
reference="permit-law:3",
|
||||
version="2026-01",
|
||||
)
|
||||
|
||||
|
||||
def evidence() -> EvidenceReference:
|
||||
return EvidenceReference(
|
||||
kind="document",
|
||||
owner_module="files",
|
||||
evidence_id="application-1",
|
||||
tenant_id="tenant-1",
|
||||
version="3",
|
||||
captured_at=NOW,
|
||||
)
|
||||
|
||||
|
||||
def case_record() -> CaseRecord:
|
||||
case_ref = ref("case", "case-1", "cases")
|
||||
context = GovernedContextEnvelope(
|
||||
tenant_id="tenant-1",
|
||||
temporal=TemporalRevision(
|
||||
revision="1",
|
||||
valid_from=NOW,
|
||||
recorded_at=NOW,
|
||||
change_reason="Case intake.",
|
||||
),
|
||||
organization_unit_ref=ref("organization_unit", "permits", "organizations"),
|
||||
function_ref=ref("function", "permit-officer", "organizations"),
|
||||
mandate_ref=ref("mandate", "permit-mandate", "mandates", version="4"),
|
||||
jurisdiction_refs=(ref("jurisdiction", "city-1", "organizations"),),
|
||||
service_ref=ref("service", "permit", "services", version="4"),
|
||||
case_ref=case_ref,
|
||||
legal_bases=(legal_basis(),),
|
||||
)
|
||||
return CaseRecord(
|
||||
reference=case_ref,
|
||||
case_number="PERMIT-2026-0001",
|
||||
case_type_key="permit-application",
|
||||
status_key="review",
|
||||
title="Permit application",
|
||||
context=context,
|
||||
service_ref=context.service_ref,
|
||||
party_refs=(ref("party", "applicant", "parties"),),
|
||||
assignment_refs=(ref("function_assignment", "assignment-1", "access"),),
|
||||
evidence_refs=(evidence(),),
|
||||
opened_at=NOW,
|
||||
recorded_at=NOW,
|
||||
change_reason="Application received.",
|
||||
)
|
||||
|
||||
|
||||
def mandate() -> MandateDefinition:
|
||||
return MandateDefinition(
|
||||
reference=ref("mandate", "permit-mandate", "mandates", version="4"),
|
||||
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=("cases.formal_decision",),
|
||||
authority_types=("permit",),
|
||||
organization_unit_refs=(ref("organization_unit", "permits", "organizations"),),
|
||||
function_refs=(ref("function", "permit-officer", "organizations"),),
|
||||
jurisdiction_refs=(ref("jurisdiction", "city-1", "organizations"),),
|
||||
legal_bases=(legal_basis(),),
|
||||
evidence=(evidence(),),
|
||||
)
|
||||
|
||||
|
||||
def command(
|
||||
*,
|
||||
key: str = "decision-1",
|
||||
result: str = "Permit granted.",
|
||||
expected_case_revision: int = 1,
|
||||
) -> CaseDecisionCommand:
|
||||
return CaseDecisionCommand(
|
||||
expected_case_revision=expected_case_revision,
|
||||
effective_at=NOW + timedelta(minutes=5),
|
||||
decision_type="permit",
|
||||
operative_result=result,
|
||||
reasoning="The submitted evidence satisfies the effective rule.",
|
||||
conditions=("Display the permit visibly.",),
|
||||
change_reason="Formal permit Decision recorded.",
|
||||
idempotency_key=key,
|
||||
)
|
||||
|
||||
|
||||
class CaseDecisionPathTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
for table in (
|
||||
CaseStatusDefinition.__table__,
|
||||
CaseTypeDefinition.__table__,
|
||||
CaseIdentity.__table__,
|
||||
CaseRecordRevision.__table__,
|
||||
CaseAccessGrant.__table__,
|
||||
CaseTimelineEntry.__table__,
|
||||
):
|
||||
table.create(self.engine)
|
||||
self.session = Session(self.engine)
|
||||
self.principal = Principal()
|
||||
upsert_case_status(
|
||||
self.session,
|
||||
self.principal,
|
||||
status_key="review",
|
||||
label="Review",
|
||||
)
|
||||
upsert_case_type(
|
||||
self.session,
|
||||
self.principal,
|
||||
type_key="permit-application",
|
||||
label="Permit application",
|
||||
initial_status_key="review",
|
||||
allowed_status_keys=("review",),
|
||||
)
|
||||
create_case(
|
||||
self.session,
|
||||
self.principal,
|
||||
record=case_record(),
|
||||
idempotency_key="case-1",
|
||||
)
|
||||
self.session.commit()
|
||||
self.decisions = Decisions()
|
||||
self.path = CaseDecisionPath(
|
||||
Registry(
|
||||
{
|
||||
CAPABILITY_ACCESS_SEMANTIC_DIRECTORY: AccessDirectory(),
|
||||
CAPABILITY_MANDATE_RESOLVER: Mandates(mandate()),
|
||||
CAPABILITY_DECISION_REGISTRY: self.decisions,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.session.close()
|
||||
self.engine.dispose()
|
||||
|
||||
def test_records_and_links_exact_decision_in_one_case_revision(self) -> None:
|
||||
result = self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(),
|
||||
)
|
||||
self.session.commit()
|
||||
|
||||
self.assertEqual(2, result.case.revision)
|
||||
self.assertEqual("Permit granted.", result.decision.operative_result)
|
||||
self.assertEqual("assignment-1", result.decision.authority_context.actor.function_assignment_ref.object_id)
|
||||
self.assertEqual("permit-mandate", result.decision.authority_context.mandate_ref.object_id)
|
||||
self.assertEqual(result.decision.reference, result.case.decision_refs[0])
|
||||
self.assertEqual(
|
||||
result.decision,
|
||||
self.path.linked(self.session, self.principal, case_id="case-1")[0],
|
||||
)
|
||||
|
||||
def test_unchanged_retry_replays_but_changed_payload_fails_closed(self) -> None:
|
||||
first = self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(),
|
||||
)
|
||||
self.session.commit()
|
||||
replay = self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(),
|
||||
)
|
||||
|
||||
self.assertTrue(replay.replayed)
|
||||
self.assertEqual(first.decision.reference, replay.decision.reference)
|
||||
with self.assertRaisesRegex(ValueError, "different Decision payload"):
|
||||
self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(result="Permit denied."),
|
||||
)
|
||||
|
||||
def test_stale_case_revision_and_missing_acting_assignment_fail_closed(self) -> None:
|
||||
self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(),
|
||||
)
|
||||
self.session.commit()
|
||||
|
||||
with self.assertRaisesRegex(CaseDecisionError, "revision conflict"):
|
||||
self.path.record(
|
||||
self.session,
|
||||
self.principal,
|
||||
case_id="case-1",
|
||||
command=command(key="another-decision"),
|
||||
)
|
||||
without_assignment = Principal(
|
||||
acting_assignment_id=None,
|
||||
function_assignment_ids=(),
|
||||
)
|
||||
current = get_case(self.session, self.principal, case_id="case-1")
|
||||
assert current is not None
|
||||
with self.assertRaisesRegex(PermissionError, "No active acting assignment"):
|
||||
self.path.record(
|
||||
self.session,
|
||||
without_assignment,
|
||||
case_id="case-1",
|
||||
command=command(
|
||||
key="third-decision",
|
||||
expected_case_revision=current.revision,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user