474 lines
15 KiB
Python
474 lines
15 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.events import EventBus, event_bus_context
|
|
from govoplan_core.core.institutional import (
|
|
EvidenceReference,
|
|
GovernedContextEnvelope,
|
|
InstitutionalReference,
|
|
ServiceBinding,
|
|
ServiceDefinition,
|
|
ServiceLaunchRequest,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_cases.backend.db.models import (
|
|
CaseAccessGrant,
|
|
CaseIdentity,
|
|
CaseRecordRevision,
|
|
CaseStatusDefinition,
|
|
CaseTimelineEntry,
|
|
CaseTypeDefinition,
|
|
)
|
|
from govoplan_cases.backend.domain import CaseGrant, CaseRecord
|
|
from govoplan_cases.backend.service import (
|
|
CaseStoreError,
|
|
case_history,
|
|
case_timeline,
|
|
create_case,
|
|
get_case,
|
|
list_cases,
|
|
update_case,
|
|
upsert_case_status,
|
|
upsert_case_type,
|
|
)
|
|
from govoplan_cases.backend.service_launcher import CaseServiceLauncher
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
scopes: frozenset[str] = frozenset()
|
|
group_ids: tuple[str, ...] = ()
|
|
function_assignment_ids: tuple[str, ...] = ()
|
|
|
|
|
|
def ref(
|
|
kind: str,
|
|
object_id: str,
|
|
owner: str,
|
|
*,
|
|
version: str | None = "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 record(*, title: str = "Permit application") -> 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.",
|
|
),
|
|
service_ref=ref("service", "permit", "services", version="4"),
|
|
case_ref=case_ref,
|
|
organization_unit_ref=ref(
|
|
"organization_unit",
|
|
"permits",
|
|
"organizations",
|
|
),
|
|
function_ref=ref("function", "case-worker", "organizations"),
|
|
)
|
|
return CaseRecord(
|
|
reference=case_ref,
|
|
case_number="PERMIT-2026-0001",
|
|
case_type_key="permit-application",
|
|
status_key="intake",
|
|
title=title,
|
|
context=context,
|
|
service_ref=context.service_ref,
|
|
party_refs=(ref("party", "applicant", "parties"),),
|
|
assignment_refs=(
|
|
ref("function_assignment", "assignment-1", "idm"),
|
|
),
|
|
evidence_refs=(
|
|
EvidenceReference(
|
|
kind="document",
|
|
owner_module="files",
|
|
evidence_id="file-1",
|
|
tenant_id="tenant-1",
|
|
version="3",
|
|
captured_at=NOW,
|
|
),
|
|
),
|
|
opened_at=NOW,
|
|
recorded_at=NOW,
|
|
deadline_at=NOW + timedelta(days=30),
|
|
change_reason="Application received.",
|
|
)
|
|
|
|
|
|
class CaseLifecycleTests(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="intake",
|
|
label="Intake",
|
|
)
|
|
upsert_case_status(
|
|
self.session,
|
|
self.principal,
|
|
status_key="review",
|
|
label="Review",
|
|
category="waiting",
|
|
)
|
|
upsert_case_status(
|
|
self.session,
|
|
self.principal,
|
|
status_key="closed",
|
|
label="Closed",
|
|
category="closed",
|
|
terminal=True,
|
|
)
|
|
upsert_case_type(
|
|
self.session,
|
|
self.principal,
|
|
type_key="permit-application",
|
|
label="Permit application",
|
|
initial_status_key="intake",
|
|
allowed_status_keys=("intake", "review", "closed"),
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_case_revisions_are_replay_safe_occ_guarded_and_event_backed(self) -> None:
|
|
events = []
|
|
bus = EventBus()
|
|
bus.subscribe("*", events.append)
|
|
with event_bus_context(bus):
|
|
created = create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(),
|
|
idempotency_key="case-create-1",
|
|
)
|
|
self.assertEqual([], events)
|
|
self.session.commit()
|
|
|
|
replay = create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(),
|
|
idempotency_key="case-create-1",
|
|
)
|
|
self.assertEqual(1, replay.revision)
|
|
|
|
with self.assertRaisesRegex(CaseStoreError, "idempotency conflict"):
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(title="Another application"),
|
|
idempotency_key="case-create-1",
|
|
)
|
|
|
|
revised = update_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"status_key": "review", "title": "Reviewed permit"},
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Review started.",
|
|
idempotency_key="case-update-1",
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(1, created.revision)
|
|
self.assertEqual(2, revised.revision)
|
|
self.assertEqual("review", revised.status_key)
|
|
self.assertEqual(
|
|
["cases.case.created", "cases.case.updated"],
|
|
[item.type for item in events],
|
|
)
|
|
self.assertEqual("case-1", events[-1].institutional_context.case_ref.object_id)
|
|
self.assertEqual(2, len(case_history(self.session, self.principal, case_id="case-1")))
|
|
timeline = case_timeline(self.session, self.principal, case_id="case-1")
|
|
self.assertEqual(2, len(timeline))
|
|
self.assertEqual(timeline[0]["event_id"], timeline[0]["audit_event_id"])
|
|
|
|
with self.assertRaisesRegex(CaseStoreError, "stale"):
|
|
update_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"title": "Stale title"},
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
change_reason="Stale update.",
|
|
idempotency_key="case-update-stale",
|
|
)
|
|
|
|
def test_terminal_state_sets_closed_time_and_list_is_server_filtered(self) -> None:
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(),
|
|
idempotency_key="case-create-2",
|
|
)
|
|
self.session.commit()
|
|
closed = update_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"status_key": "closed"},
|
|
recorded_at=NOW + timedelta(hours=1),
|
|
change_reason="Decision became final.",
|
|
idempotency_key="case-close-1",
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(NOW + timedelta(hours=1), closed.closed_at)
|
|
items, total = list_cases(
|
|
self.session,
|
|
self.principal,
|
|
status_keys=("closed",),
|
|
query="permit-2026",
|
|
)
|
|
self.assertEqual(1, total)
|
|
self.assertEqual("case-1", items[0].reference.object_id)
|
|
|
|
def test_tenant_boundary_and_catalog_revision_conflicts_fail_closed(self) -> None:
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(),
|
|
idempotency_key="case-create-3",
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertIsNone(
|
|
get_case(
|
|
self.session,
|
|
Principal(tenant_id="tenant-2"),
|
|
case_id="case-1",
|
|
)
|
|
)
|
|
with self.assertRaisesRegex(CaseStoreError, "stale"):
|
|
upsert_case_status(
|
|
self.session,
|
|
self.principal,
|
|
status_key="review",
|
|
label="In review",
|
|
expected_revision=99,
|
|
)
|
|
with self.assertRaisesRegex(CaseStoreError, "cannot be deactivated"):
|
|
upsert_case_type(
|
|
self.session,
|
|
self.principal,
|
|
type_key="permit-application",
|
|
label="Permit application",
|
|
initial_status_key="intake",
|
|
allowed_status_keys=("intake", "review", "closed"),
|
|
active=False,
|
|
expected_revision=1,
|
|
)
|
|
|
|
def test_restricted_cases_filter_reads_and_honor_explicit_permissions(self) -> None:
|
|
restricted = record()
|
|
restricted = CaseRecord.from_mapping(
|
|
{
|
|
**restricted.to_dict(),
|
|
"access_mode": "restricted",
|
|
"access_grants": [
|
|
CaseGrant(
|
|
subject_kind="group",
|
|
subject_id="reviewers",
|
|
permissions=("read",),
|
|
).to_dict()
|
|
],
|
|
}
|
|
)
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=restricted,
|
|
idempotency_key="case-restricted-create",
|
|
)
|
|
self.session.commit()
|
|
|
|
outsider = Principal(account_id="account-2")
|
|
self.assertIsNone(get_case(self.session, outsider, case_id="case-1"))
|
|
self.assertEqual(0, list_cases(self.session, outsider)[1])
|
|
self.assertEqual((), case_history(self.session, outsider, case_id="case-1"))
|
|
self.assertEqual((), case_timeline(self.session, outsider, case_id="case-1"))
|
|
|
|
reader = Principal(account_id="account-3", group_ids=("reviewers",))
|
|
self.assertIsNotNone(get_case(self.session, reader, case_id="case-1"))
|
|
self.assertEqual(1, list_cases(self.session, reader)[1])
|
|
with self.assertRaises(PermissionError):
|
|
update_case(
|
|
self.session,
|
|
reader,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"title": "Unauthorized title"},
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Attempted update.",
|
|
idempotency_key="case-reader-update",
|
|
)
|
|
|
|
def test_assignment_grants_and_access_revision_are_effective(self) -> None:
|
|
restricted = CaseRecord.from_mapping(
|
|
{
|
|
**record().to_dict(),
|
|
"access_mode": "restricted",
|
|
}
|
|
)
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=restricted,
|
|
idempotency_key="case-assignment-create",
|
|
)
|
|
self.session.commit()
|
|
|
|
assignee = Principal(
|
|
account_id="account-4",
|
|
function_assignment_ids=("assignment-1",),
|
|
)
|
|
revised = update_case(
|
|
self.session,
|
|
assignee,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"title": "Assigned case"},
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Assigned worker updated the case.",
|
|
idempotency_key="case-assignee-update",
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(2, revised.revision)
|
|
|
|
shared = update_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id="case-1",
|
|
expected_revision=2,
|
|
changes={
|
|
"access_grants": [
|
|
CaseGrant(
|
|
subject_kind="account",
|
|
subject_id="account-5",
|
|
permissions=("read", "update"),
|
|
)
|
|
]
|
|
},
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
change_reason="Granted direct collaboration access.",
|
|
idempotency_key="case-share-update",
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(3, shared.revision)
|
|
self.assertIsNotNone(
|
|
get_case(self.session, Principal(account_id="account-5"), case_id="case-1")
|
|
)
|
|
self.assertEqual(3, len(case_history(self.session, self.principal, case_id="case-1")))
|
|
|
|
def test_malformed_access_grant_is_rejected(self) -> None:
|
|
create_case(
|
|
self.session,
|
|
self.principal,
|
|
record=record(),
|
|
idempotency_key="case-malformed-create",
|
|
)
|
|
self.session.commit()
|
|
with self.assertRaisesRegex(CaseStoreError, "must be objects"):
|
|
update_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id="case-1",
|
|
expected_revision=1,
|
|
changes={"access_grants": ["account-2"]},
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Invalid share.",
|
|
idempotency_key="case-malformed-share",
|
|
)
|
|
|
|
def test_exact_service_launch_creates_and_replays_one_case(self) -> None:
|
|
definition = ServiceDefinition(
|
|
reference=ref("service", "permit", "services", version="4"),
|
|
key="permit.apply",
|
|
temporal=TemporalRevision(
|
|
revision="4",
|
|
valid_from=NOW - timedelta(days=1),
|
|
recorded_at=NOW - timedelta(days=2),
|
|
change_reason="Published permit service.",
|
|
),
|
|
title="Apply for a permit",
|
|
audience=("authenticated",),
|
|
bindings=(ServiceBinding("case", "permit-application"),),
|
|
publication_state="published",
|
|
)
|
|
request = ServiceLaunchRequest(
|
|
service_ref=definition.reference,
|
|
binding=definition.bindings[0],
|
|
idempotency_key="portal-launch-1",
|
|
requested_at=NOW,
|
|
parameters={},
|
|
)
|
|
|
|
first = CaseServiceLauncher().launch_service(
|
|
self.session,
|
|
self.principal,
|
|
definition=definition,
|
|
request=request,
|
|
)
|
|
self.session.commit()
|
|
second = CaseServiceLauncher().launch_service(
|
|
self.session,
|
|
self.principal,
|
|
definition=definition,
|
|
request=request,
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(first.target_ref, second.target_ref)
|
|
self.assertFalse(first.replayed)
|
|
self.assertTrue(second.replayed)
|
|
self.assertEqual(1, list_cases(self.session, self.principal)[1])
|
|
self.assertEqual("4", get_case(
|
|
self.session,
|
|
self.principal,
|
|
case_id=first.target_ref.object_id,
|
|
).service_ref.version)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|