351 lines
11 KiB
Python
351 lines
11 KiB
Python
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()
|