1347 lines
48 KiB
Python
1347 lines
48 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.institutional import (
|
|
CAPABILITY_FORM_DEFINITIONS,
|
|
CAPABILITY_SERVICE_DEFINITIONS,
|
|
FormConditionExpression,
|
|
FormDefinition,
|
|
FormFieldDefinition,
|
|
InstitutionalReference,
|
|
ServiceBinding,
|
|
ServiceDefinition,
|
|
ServiceLaunchRequest,
|
|
ServiceLaunchResult,
|
|
TemporalRevision,
|
|
)
|
|
from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH
|
|
from govoplan_forms.backend.db.models import FormDefinitionRevision
|
|
from govoplan_forms.backend.service import (
|
|
SqlFormDefinitionProvider,
|
|
record_form_definition,
|
|
)
|
|
from govoplan_forms_runtime.backend.db.models import (
|
|
FormAcknowledgement,
|
|
FormAssistedConfirmation,
|
|
FormHandoffEffect,
|
|
FormInstanceEvent,
|
|
FormInstanceIdentity,
|
|
FormInstanceRevision,
|
|
FormIntakeProfile,
|
|
FormIntakeSession,
|
|
FormStatusAccessGrant,
|
|
FormStatusAccessPolicy,
|
|
FormStatusAccessToken,
|
|
)
|
|
from govoplan_forms_runtime.backend.service import (
|
|
FormRuntimeError,
|
|
FormRuntimeService,
|
|
FormsServiceLauncher,
|
|
)
|
|
from govoplan_forms_runtime.backend.handoffs import FormHandoffService
|
|
from govoplan_forms_runtime.backend.intake import FormIntakeError, FormIntakeService
|
|
from govoplan_forms_runtime.backend.status_access import (
|
|
FormStatusAccessError,
|
|
FormStatusAccessService,
|
|
FormStatusUnavailable,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
account_id: str = "account-1"
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, provider: object, **capabilities: object) -> None:
|
|
self.capabilities = {
|
|
CAPABILITY_FORM_DEFINITIONS: provider,
|
|
**capabilities,
|
|
}
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name in self.capabilities
|
|
|
|
def require_capability(self, name: str) -> object:
|
|
return self.capabilities[name]
|
|
|
|
def capability(self, name: str) -> object:
|
|
return self.capabilities[name]
|
|
|
|
|
|
class NotificationProvider:
|
|
def __init__(self) -> None:
|
|
self.requests: list[object] = []
|
|
|
|
def tenant_id_for_notification(
|
|
self,
|
|
session: object,
|
|
*,
|
|
notification_id: str,
|
|
) -> str | None:
|
|
del session, notification_id
|
|
return "tenant-1"
|
|
|
|
def enqueue_notification(
|
|
self,
|
|
session: object,
|
|
request: object,
|
|
*,
|
|
enqueue_delivery: bool = True,
|
|
) -> dict[str, object]:
|
|
del session, enqueue_delivery
|
|
self.requests.append(request)
|
|
return {"id": f"notification-{len(self.requests)}"}
|
|
|
|
def deliver_notification(
|
|
self,
|
|
session: object,
|
|
*,
|
|
notification_id: str,
|
|
) -> dict[str, object]:
|
|
del session, notification_id
|
|
return {}
|
|
|
|
def deliver_pending(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
limit: int = 50,
|
|
) -> dict[str, object]:
|
|
del session, tenant_id, limit
|
|
return {}
|
|
|
|
|
|
def form_definition(
|
|
*,
|
|
form_id: str = "permit-form",
|
|
revision: str = "1",
|
|
policy_refs: tuple[str, ...] = (),
|
|
) -> FormDefinition:
|
|
return FormDefinition(
|
|
reference=InstitutionalReference(
|
|
kind="form",
|
|
owner_module="forms",
|
|
object_id=form_id,
|
|
tenant_id="tenant-1",
|
|
version=revision,
|
|
),
|
|
key=form_id,
|
|
temporal=TemporalRevision(
|
|
revision=revision,
|
|
recorded_at=NOW + timedelta(minutes=int(revision) - 2),
|
|
change_reason=("Initial schema." if revision == "1" else "Revise schema."),
|
|
),
|
|
title="Permit form",
|
|
fields=(
|
|
FormFieldDefinition(
|
|
key="name",
|
|
label="Name",
|
|
required=True,
|
|
constraints={"min_length": 2},
|
|
),
|
|
FormFieldDefinition(
|
|
key="delivery",
|
|
label="Delivery",
|
|
value_type="choice",
|
|
options=("portal", "mail"),
|
|
),
|
|
FormFieldDefinition(
|
|
key="email",
|
|
label="Email",
|
|
value_type="email",
|
|
constraints={"max_length": 320},
|
|
),
|
|
),
|
|
publication_state="published",
|
|
allow_drafts=True,
|
|
handoff_kinds=("case",),
|
|
policy_refs=policy_refs,
|
|
)
|
|
|
|
|
|
class FormsRuntimeTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
for table in (
|
|
FormDefinitionRevision.__table__,
|
|
FormInstanceIdentity.__table__,
|
|
FormInstanceRevision.__table__,
|
|
FormInstanceEvent.__table__,
|
|
FormHandoffEffect.__table__,
|
|
FormIntakeProfile.__table__,
|
|
FormIntakeSession.__table__,
|
|
FormAssistedConfirmation.__table__,
|
|
FormStatusAccessPolicy.__table__,
|
|
FormStatusAccessGrant.__table__,
|
|
FormStatusAccessToken.__table__,
|
|
FormAcknowledgement.__table__,
|
|
):
|
|
table.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
self.principal = Principal()
|
|
self.definition = record_form_definition(
|
|
self.session,
|
|
self.principal,
|
|
definition=form_definition(),
|
|
)
|
|
self.registry = Registry(SqlFormDefinitionProvider())
|
|
self.runtime = FormRuntimeService(self.registry)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_draft_submit_occ_replay_and_status_history(self) -> None:
|
|
draft = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={},
|
|
idempotency_key="create-1",
|
|
recorded_at=NOW,
|
|
)
|
|
self.assertEqual("draft", draft.status)
|
|
self.assertEqual("warning", draft.validation_results[0]["severity"])
|
|
|
|
saved = self.runtime.update_draft(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "Ada", "delivery": "portal"},
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="save-1",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
change_reason="Complete required values.",
|
|
)
|
|
submitted = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=2,
|
|
values=saved.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="submit-1",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
replay = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=2,
|
|
values=saved.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="submit-1",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
|
|
self.assertEqual("submitted", submitted.status)
|
|
self.assertIsNotNone(submitted.receipt_id)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(
|
|
[3, 2, 1],
|
|
[
|
|
item.revision
|
|
for item in self.runtime.history(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
)
|
|
],
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "stale"):
|
|
self.runtime.transition_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=2,
|
|
status="validated",
|
|
idempotency_key="transition-stale",
|
|
recorded_at=NOW + timedelta(minutes=3),
|
|
change_reason="Review complete.",
|
|
)
|
|
|
|
def test_validation_policy_tenant_and_handoff_fail_closed(self) -> None:
|
|
draft = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={},
|
|
idempotency_key="create-2",
|
|
recorded_at=NOW,
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "failed validation"):
|
|
self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "A"},
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="invalid-submit",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
with self.assertRaisesRegex(PermissionError, "denied"):
|
|
self.runtime.get_instance(
|
|
self.session,
|
|
Principal(account_id="account-2"),
|
|
instance_id=draft.instance_id,
|
|
)
|
|
|
|
submitted = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "Ada"},
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="valid-submit",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "cross tenants"):
|
|
self.runtime.handoff_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=submitted.instance_id,
|
|
expected_revision=2,
|
|
target_ref=InstitutionalReference(
|
|
kind="case",
|
|
owner_module="cases",
|
|
object_id="case-1",
|
|
tenant_id="tenant-2",
|
|
),
|
|
idempotency_key="handoff-invalid",
|
|
recorded_at=NOW + timedelta(minutes=3),
|
|
change_reason="Create case.",
|
|
)
|
|
current = self.runtime.get_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
)
|
|
self.assertEqual(2, current.revision if current else None)
|
|
|
|
protected = record_form_definition(
|
|
self.session,
|
|
self.principal,
|
|
definition=form_definition(
|
|
form_id="protected-form",
|
|
policy_refs=("policy:protected-intake",),
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(PermissionError, "policy evaluator"):
|
|
self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=protected.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="protected-start",
|
|
recorded_at=NOW + timedelta(minutes=4),
|
|
)
|
|
|
|
def test_service_launcher_retains_exact_service_form_and_replay(self) -> None:
|
|
binding = ServiceBinding(kind="form", reference="permit-form/1")
|
|
service = ServiceDefinition(
|
|
reference=InstitutionalReference(
|
|
kind="service",
|
|
owner_module="services",
|
|
object_id="permit-service",
|
|
tenant_id="tenant-1",
|
|
version="4",
|
|
),
|
|
key="permit-service",
|
|
temporal=TemporalRevision(
|
|
revision="4",
|
|
recorded_at=NOW - timedelta(minutes=2),
|
|
change_reason="Publish form entry.",
|
|
),
|
|
title="Apply for permit",
|
|
audience=("resident",),
|
|
bindings=(binding,),
|
|
publication_state="published",
|
|
)
|
|
request = ServiceLaunchRequest(
|
|
service_ref=service.reference,
|
|
binding=binding,
|
|
idempotency_key="portal-1",
|
|
requested_at=NOW,
|
|
parameters={},
|
|
)
|
|
launcher = FormsServiceLauncher(self.registry)
|
|
first = launcher.launch_service(
|
|
self.session,
|
|
self.principal,
|
|
definition=service,
|
|
request=request,
|
|
)
|
|
replay = launcher.launch_service(
|
|
self.session,
|
|
self.principal,
|
|
definition=service,
|
|
request=request,
|
|
)
|
|
|
|
self.assertEqual(
|
|
"form_submission", first.target_ref.kind if first.target_ref else None
|
|
)
|
|
self.assertEqual("1", first.metadata["form_definition_revision"])
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(first.target_ref.object_id, replay.target_ref.object_id)
|
|
|
|
def test_create_replay_is_actor_bound_and_survives_schema_supersession(
|
|
self,
|
|
) -> None:
|
|
first = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="create-replay",
|
|
recorded_at=NOW,
|
|
)
|
|
record_form_definition(
|
|
self.session,
|
|
self.principal,
|
|
definition=form_definition(revision="2"),
|
|
expected_revision="1",
|
|
)
|
|
|
|
replay = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="create-replay",
|
|
recorded_at=NOW,
|
|
)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(first.instance_id, replay.instance_id)
|
|
|
|
with self.assertRaisesRegex(FormRuntimeError, "another actor"):
|
|
self.runtime.create_instance(
|
|
self.session,
|
|
Principal(account_id="account-2"),
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="create-replay",
|
|
recorded_at=NOW,
|
|
)
|
|
|
|
def test_runtime_rejects_a_provider_returning_another_exact_revision(self) -> None:
|
|
class MismatchedProvider:
|
|
def get_form_definition(
|
|
self,
|
|
session,
|
|
principal,
|
|
*,
|
|
reference,
|
|
effective_at=None,
|
|
):
|
|
return form_definition(revision="2")
|
|
|
|
def list_form_definitions(
|
|
self,
|
|
session,
|
|
principal,
|
|
*,
|
|
tenant_id,
|
|
query="",
|
|
limit=100,
|
|
):
|
|
return ()
|
|
|
|
runtime = FormRuntimeService(Registry(MismatchedProvider()))
|
|
with self.assertRaisesRegex(FormRuntimeError, "different definition"):
|
|
runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="mismatched-provider",
|
|
recorded_at=NOW,
|
|
)
|
|
|
|
def test_client_supplied_instance_id_cannot_replace_existing_state(self) -> None:
|
|
self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="fixed-instance-first",
|
|
recorded_at=NOW,
|
|
instance_id="fixed-instance",
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "instance id is already"):
|
|
self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Grace"},
|
|
idempotency_key="fixed-instance-second",
|
|
recorded_at=NOW,
|
|
instance_id="fixed-instance",
|
|
)
|
|
|
|
def test_hidden_required_fields_are_not_required_or_persisted(self) -> None:
|
|
conditional = replace(
|
|
form_definition(form_id="conditional-form"),
|
|
fields=(
|
|
FormFieldDefinition(
|
|
key="include_details",
|
|
label="Include details",
|
|
value_type="boolean",
|
|
),
|
|
FormFieldDefinition(
|
|
key="details",
|
|
label="Details",
|
|
required=True,
|
|
visibility_condition=FormConditionExpression(
|
|
kind="predicate",
|
|
field_key="include_details",
|
|
operator="eq",
|
|
value=True,
|
|
),
|
|
),
|
|
),
|
|
)
|
|
stored = record_form_definition(
|
|
self.session,
|
|
self.principal,
|
|
definition=conditional,
|
|
)
|
|
hidden = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=stored.reference,
|
|
values={"include_details": False, "details": "stale secret"},
|
|
idempotency_key="conditional-hidden",
|
|
recorded_at=NOW,
|
|
)
|
|
self.assertNotIn("details", hidden.values)
|
|
submitted = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=hidden.instance_id,
|
|
expected_revision=1,
|
|
values=hidden.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="conditional-submit",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertEqual("submitted", submitted.status)
|
|
|
|
shown = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=stored.reference,
|
|
values={"include_details": True},
|
|
idempotency_key="conditional-shown",
|
|
recorded_at=NOW,
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "required"):
|
|
self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=shown.instance_id,
|
|
expected_revision=1,
|
|
values=shown.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="conditional-shown-submit",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
|
|
def test_native_handoff_is_durable_replay_safe_and_reconciles_unknown(self) -> None:
|
|
form_binding = ServiceBinding(kind="form", reference="permit-form/1")
|
|
case_binding = ServiceBinding(kind="case", reference="permit-case")
|
|
service = ServiceDefinition(
|
|
reference=InstitutionalReference(
|
|
kind="service",
|
|
owner_module="services",
|
|
object_id="permit-service",
|
|
tenant_id="tenant-1",
|
|
version="4",
|
|
),
|
|
key="permit-service",
|
|
temporal=TemporalRevision(
|
|
revision="4",
|
|
recorded_at=NOW - timedelta(minutes=2),
|
|
change_reason="Publish intake targets.",
|
|
),
|
|
title="Apply for permit",
|
|
audience=("resident",),
|
|
bindings=(form_binding, case_binding),
|
|
publication_state="published",
|
|
)
|
|
|
|
class Services:
|
|
def get_service_definition(
|
|
self, session, principal, *, reference, effective_at=None
|
|
):
|
|
return service if reference == service.reference else None
|
|
|
|
def list_service_definitions(
|
|
self, session, principal, *, tenant_id, query="", limit=100
|
|
):
|
|
return (service,)
|
|
|
|
class Cases:
|
|
fail_unknown = False
|
|
|
|
def launch_service(self, session, principal, *, definition, request):
|
|
if self.fail_unknown:
|
|
raise ConnectionError("Provider response was lost.")
|
|
return ServiceLaunchResult(
|
|
service_ref=definition.reference,
|
|
binding=request.binding,
|
|
state="started",
|
|
target_ref=InstitutionalReference(
|
|
kind="case",
|
|
owner_module="cases",
|
|
object_id="case-1",
|
|
tenant_id="tenant-1",
|
|
version="1",
|
|
),
|
|
href="/cases/case-1",
|
|
replayed=request.idempotency_key.startswith("known"),
|
|
metadata={"case_number": "CASE-1"},
|
|
)
|
|
|
|
cases = Cases()
|
|
registry = Registry(
|
|
SqlFormDefinitionProvider(),
|
|
**{
|
|
CAPABILITY_SERVICE_DEFINITIONS: Services(),
|
|
"cases.service_launcher": cases,
|
|
},
|
|
)
|
|
runtime = FormRuntimeService(registry)
|
|
draft = runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="native-create",
|
|
recorded_at=NOW,
|
|
service_ref=service.reference,
|
|
service_binding=form_binding,
|
|
)
|
|
submitted = runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
values=draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="native-submit",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
handoffs = FormHandoffService(registry)
|
|
prepared = handoffs.prepare(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=submitted.instance_id,
|
|
expected_revision=2,
|
|
binding_kind="case",
|
|
binding_reference="permit-case",
|
|
idempotency_key="native-handoff",
|
|
requested_at=NOW + timedelta(minutes=2),
|
|
allow_all=True,
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual("requested", prepared.state)
|
|
self.assertEqual(
|
|
"requested",
|
|
self.session.query(FormHandoffEffect).one().state,
|
|
)
|
|
accepted, revised = handoffs.execute(
|
|
self.session,
|
|
self.principal,
|
|
effect_id=prepared.effect_id,
|
|
executed_at=NOW + timedelta(minutes=2),
|
|
allow_all=True,
|
|
)
|
|
self.assertEqual("accepted", accepted.state)
|
|
self.assertEqual("handed_off", revised.status)
|
|
replay = handoffs.prepare(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=submitted.instance_id,
|
|
expected_revision=2,
|
|
binding_kind="case",
|
|
binding_reference="permit-case",
|
|
idempotency_key="native-handoff",
|
|
requested_at=NOW + timedelta(minutes=2),
|
|
allow_all=True,
|
|
)
|
|
self.assertEqual("accepted", replay.state)
|
|
|
|
second = runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Grace"},
|
|
idempotency_key="unknown-create",
|
|
recorded_at=NOW,
|
|
service_ref=service.reference,
|
|
service_binding=form_binding,
|
|
)
|
|
second = runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=second.instance_id,
|
|
expected_revision=1,
|
|
values=second.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="unknown-submit",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
unknown = handoffs.prepare(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=second.instance_id,
|
|
expected_revision=2,
|
|
binding_kind="case",
|
|
binding_reference=None,
|
|
idempotency_key="unknown-handoff",
|
|
requested_at=NOW + timedelta(minutes=2),
|
|
allow_all=True,
|
|
)
|
|
self.session.commit()
|
|
cases.fail_unknown = True
|
|
unknown, _ = handoffs.execute(
|
|
self.session,
|
|
self.principal,
|
|
effect_id=unknown.effect_id,
|
|
executed_at=NOW + timedelta(minutes=2),
|
|
allow_all=True,
|
|
)
|
|
self.assertEqual("outcome_unknown", unknown.state)
|
|
with self.assertRaisesRegex(FormRuntimeError, "must be reconciled"):
|
|
handoffs.execute(
|
|
self.session,
|
|
self.principal,
|
|
effect_id=unknown.effect_id,
|
|
executed_at=NOW + timedelta(minutes=3),
|
|
allow_all=True,
|
|
)
|
|
cases.fail_unknown = False
|
|
reconciled, revised = handoffs.execute(
|
|
self.session,
|
|
self.principal,
|
|
effect_id=unknown.effect_id,
|
|
executed_at=NOW + timedelta(minutes=4),
|
|
allow_all=True,
|
|
reconcile=True,
|
|
)
|
|
self.assertEqual("reconciled", reconciled.state)
|
|
self.assertEqual("handed_off", revised.status)
|
|
|
|
def test_anonymous_and_invitation_intake_are_hashed_expiring_and_replay_safe(
|
|
self,
|
|
) -> None:
|
|
intake = FormIntakeService(self.registry)
|
|
available = intake.list_available_definitions(
|
|
self.session,
|
|
self.principal,
|
|
)
|
|
self.assertEqual(
|
|
(self.definition.reference,), tuple(item.reference for item in available)
|
|
)
|
|
anonymous = intake.create_profile(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="anonymous",
|
|
custodian_ref="user:account-1",
|
|
rate_limit_per_minute=10,
|
|
recorded_at=NOW,
|
|
)
|
|
started = intake.start_anonymous(
|
|
self.session,
|
|
public_id=anonymous.public_id,
|
|
values={"name": "Ada"},
|
|
idempotency_key="anonymous-start",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertIsNotNone(started.token)
|
|
stored_session = self.session.query(FormIntakeSession).one()
|
|
self.assertNotEqual(started.token, stored_session.token_sha256)
|
|
current, definition = intake.get_public_instance(
|
|
self.session,
|
|
token=str(started.token),
|
|
now=NOW + timedelta(minutes=2),
|
|
)
|
|
self.assertEqual("permit-form", definition.reference.object_id)
|
|
submitted = intake.submit_public(
|
|
self.session,
|
|
token=str(started.token),
|
|
expected_revision=current.revision,
|
|
values=current.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="anonymous-submit",
|
|
recorded_at=NOW + timedelta(minutes=3),
|
|
)
|
|
self.assertEqual("submitted", submitted.status)
|
|
|
|
replay = intake.start_anonymous(
|
|
self.session,
|
|
public_id=anonymous.public_id,
|
|
values={"name": "Ada"},
|
|
idempotency_key="anonymous-start",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertIsNone(replay.token)
|
|
self.assertEqual(started.instance.instance_id, replay.instance.instance_id)
|
|
|
|
invitation = intake.create_profile(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="invitation",
|
|
custodian_ref="user:account-1",
|
|
recorded_at=NOW,
|
|
)
|
|
issued = intake.issue_invitation(
|
|
self.session,
|
|
self.principal,
|
|
profile_id=invitation.profile_id,
|
|
idempotency_key="invite-1",
|
|
recorded_at=NOW,
|
|
)
|
|
invited = intake.start_invitation(
|
|
self.session,
|
|
token=str(issued.token),
|
|
values={"name": "Grace"},
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertEqual("active", invited.status)
|
|
self.assertEqual("invitation", invited.mode)
|
|
|
|
with self.assertRaisesRegex(FormIntakeError, "unavailable"):
|
|
intake.get_public_instance(
|
|
self.session,
|
|
token="x" * 40,
|
|
now=NOW + timedelta(minutes=1),
|
|
)
|
|
|
|
def test_assisted_intake_is_resumable_and_requires_exact_readback_evidence(
|
|
self,
|
|
) -> None:
|
|
intake = FormIntakeService(self.registry)
|
|
profile = intake.create_profile(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="assisted",
|
|
custodian_ref="function:parking-permits",
|
|
recorded_at=NOW,
|
|
)
|
|
started = intake.start_assisted(
|
|
self.session,
|
|
self.principal,
|
|
profile_id=profile.profile_id,
|
|
values={"name": "Ada", "delivery": "mail"},
|
|
channel="counter",
|
|
affected_party_ref="party:resident-1",
|
|
represented_party_ref=None,
|
|
authority_basis="self",
|
|
purpose="Apply for a resident parking permit.",
|
|
legal_basis_ref="law:resident-parking",
|
|
consent_basis="verbal-consent",
|
|
notice_given=True,
|
|
responsible_function_ref="function:parking-permits",
|
|
language="de",
|
|
accessibility_needs=("plain-language",),
|
|
field_sources={
|
|
"name": {
|
|
"source": "person_statement",
|
|
"confidence": "stated",
|
|
"declared_by_ref": "party:resident-1",
|
|
},
|
|
"delivery": {
|
|
"source": "person_statement",
|
|
"confidence": "stated",
|
|
"declared_by_ref": "party:resident-1",
|
|
},
|
|
},
|
|
idempotency_key="assisted-start-1",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertIsNone(started.token)
|
|
self.assertEqual("assisted", started.mode)
|
|
self.assertEqual("counter", started.instance.metadata["intake"]["channel"])
|
|
replay = intake.start_assisted(
|
|
self.session,
|
|
self.principal,
|
|
profile_id=profile.profile_id,
|
|
values={"name": "Ada", "delivery": "mail"},
|
|
channel="counter",
|
|
affected_party_ref="party:resident-1",
|
|
represented_party_ref=None,
|
|
authority_basis="self",
|
|
purpose="Apply for a resident parking permit.",
|
|
legal_basis_ref="law:resident-parking",
|
|
consent_basis="verbal-consent",
|
|
notice_given=True,
|
|
responsible_function_ref="function:parking-permits",
|
|
language="de",
|
|
accessibility_needs=("plain-language",),
|
|
field_sources={
|
|
"name": {
|
|
"source": "person_statement",
|
|
"confidence": "stated",
|
|
"declared_by_ref": "party:resident-1",
|
|
},
|
|
"delivery": {
|
|
"source": "person_statement",
|
|
"confidence": "stated",
|
|
"declared_by_ref": "party:resident-1",
|
|
},
|
|
},
|
|
idempotency_key="assisted-start-1",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(started.instance.instance_id, replay.instance.instance_id)
|
|
|
|
with self.assertRaisesRegex(FormRuntimeError, "read-back confirmation"):
|
|
self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=started.instance.instance_id,
|
|
expected_revision=1,
|
|
values=started.instance.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="assisted-submit-unconfirmed",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
with self.assertRaisesRegex(FormIntakeError, "Save assisted corrections"):
|
|
intake.record_assisted_confirmation(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=started.instance.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "Ada Lovelace", "delivery": "mail"},
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
outcome="corrected",
|
|
method="spoken_readback",
|
|
confirmed_by_ref="party:resident-1",
|
|
confirmed_at=NOW + timedelta(minutes=2),
|
|
idempotency_key="readback-unsaved",
|
|
field_sources={
|
|
key: {"source": "person_statement", "confidence": "stated"}
|
|
for key in ("name", "delivery")
|
|
},
|
|
correction_note="Applicant corrected the full name.",
|
|
)
|
|
confirmed = intake.record_assisted_confirmation(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=started.instance.instance_id,
|
|
expected_revision=1,
|
|
values=started.instance.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
outcome="confirmed",
|
|
method="spoken_readback",
|
|
confirmed_by_ref="party:resident-1",
|
|
confirmed_at=NOW + timedelta(minutes=3),
|
|
idempotency_key="readback-1",
|
|
field_sources={
|
|
key: {"source": "person_statement", "confidence": "stated"}
|
|
for key in ("name", "delivery")
|
|
},
|
|
)
|
|
self.assertEqual("confirmed", confirmed.outcome)
|
|
|
|
corrected_draft = self.runtime.update_draft(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=started.instance.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "Ada Lovelace", "delivery": "mail"},
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="assisted-correction-save",
|
|
recorded_at=NOW + timedelta(minutes=4),
|
|
change_reason="Correct the applicant name after read-back.",
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "read-back confirmation"):
|
|
self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=corrected_draft.instance_id,
|
|
expected_revision=2,
|
|
values=corrected_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="assisted-submit-stale-readback",
|
|
recorded_at=NOW + timedelta(minutes=5),
|
|
)
|
|
corrected = intake.record_assisted_confirmation(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=corrected_draft.instance_id,
|
|
expected_revision=2,
|
|
values=corrected_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
outcome="corrected",
|
|
method="written_preview",
|
|
confirmed_by_ref="party:resident-1",
|
|
confirmed_at=NOW + timedelta(minutes=6),
|
|
idempotency_key="readback-2",
|
|
field_sources={
|
|
key: {"source": "person_statement", "confidence": "stated"}
|
|
for key in ("name", "delivery")
|
|
},
|
|
correction_note="Applicant corrected the full name.",
|
|
)
|
|
self.assertNotEqual(confirmed.payload_sha256, corrected.payload_sha256)
|
|
submitted = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=corrected_draft.instance_id,
|
|
expected_revision=2,
|
|
values=corrected_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="assisted-submit-confirmed",
|
|
recorded_at=NOW + timedelta(minutes=7),
|
|
)
|
|
self.assertEqual("submitted", submitted.status)
|
|
stored_session = (
|
|
self.session.query(FormIntakeSession)
|
|
.filter(FormIntakeSession.instance_id == submitted.instance_id)
|
|
.one()
|
|
)
|
|
self.assertEqual("submitted", stored_session.status)
|
|
self.assertEqual(
|
|
2,
|
|
len(
|
|
intake.list_assisted_confirmations(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=submitted.instance_id,
|
|
)
|
|
),
|
|
)
|
|
|
|
def test_status_access_modes_are_configurable_and_public_projection_is_bounded(
|
|
self,
|
|
) -> None:
|
|
notifications = NotificationProvider()
|
|
registry = Registry(
|
|
SqlFormDefinitionProvider(),
|
|
**{CAPABILITY_NOTIFICATIONS_DISPATCH: notifications},
|
|
)
|
|
runtime = FormRuntimeService(registry)
|
|
status_access = FormStatusAccessService(registry)
|
|
|
|
policy = status_access.upsert_policy(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="authenticated",
|
|
enabled=True,
|
|
email_field_key=None,
|
|
recorded_at=NOW,
|
|
)
|
|
authenticated_draft = runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada", "delivery": "portal"},
|
|
idempotency_key="status-auth-create",
|
|
recorded_at=NOW + timedelta(minutes=1),
|
|
)
|
|
authenticated_submission = runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=authenticated_draft.instance_id,
|
|
expected_revision=1,
|
|
values=authenticated_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="status-auth-submit",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
authenticated_summary = status_access.access_summary_for_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=authenticated_submission.instance_id,
|
|
)
|
|
self.assertEqual("authenticated", authenticated_summary["mode"])
|
|
authenticated_projection = status_access.get_authenticated_projection(
|
|
self.session,
|
|
self.principal,
|
|
tracking_id=authenticated_summary["tracking_id"],
|
|
observed_at=NOW + timedelta(minutes=3),
|
|
)
|
|
self.assertEqual("submitted", authenticated_projection["status"])
|
|
self.assertNotIn("values", authenticated_projection)
|
|
self.assertNotIn("actor_id", authenticated_projection)
|
|
with self.assertRaises(FormStatusUnavailable):
|
|
status_access.get_authenticated_projection(
|
|
self.session,
|
|
Principal(account_id="another-account"),
|
|
tracking_id=authenticated_summary["tracking_id"],
|
|
observed_at=NOW + timedelta(minutes=3),
|
|
)
|
|
|
|
with self.assertRaisesRegex(FormStatusAccessError, "email value type"):
|
|
status_access.upsert_policy(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="email_link",
|
|
enabled=True,
|
|
email_field_key="name",
|
|
expected_revision=policy.revision,
|
|
recorded_at=NOW + timedelta(minutes=4),
|
|
)
|
|
|
|
policy = status_access.upsert_policy(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="email_link",
|
|
enabled=True,
|
|
email_field_key="email",
|
|
token_ttl_seconds=900,
|
|
request_limit_per_hour=3,
|
|
expected_revision=policy.revision,
|
|
recorded_at=NOW + timedelta(minutes=4),
|
|
)
|
|
email_draft = runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={
|
|
"name": "Ada",
|
|
"delivery": "mail",
|
|
"email": "Ada@example.test",
|
|
},
|
|
idempotency_key="status-email-create",
|
|
recorded_at=NOW + timedelta(minutes=5),
|
|
)
|
|
email_submission = runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=email_draft.instance_id,
|
|
expected_revision=1,
|
|
values=email_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="status-email-submit",
|
|
recorded_at=NOW + timedelta(minutes=6),
|
|
)
|
|
email_summary = status_access.access_summary_for_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=email_submission.instance_id,
|
|
)
|
|
self.assertFalse(
|
|
status_access.request_email_link(
|
|
self.session,
|
|
tracking_id=email_summary["tracking_id"],
|
|
email="wrong@example.test",
|
|
requested_at=NOW + timedelta(minutes=7),
|
|
)
|
|
)
|
|
self.assertEqual([], notifications.requests)
|
|
self.assertTrue(
|
|
status_access.request_email_link(
|
|
self.session,
|
|
tracking_id=email_summary["tracking_id"],
|
|
email="ada@EXAMPLE.test",
|
|
requested_at=NOW + timedelta(minutes=8),
|
|
)
|
|
)
|
|
first_url = notifications.requests[-1].action_url
|
|
first_token = parse_qs(urlsplit(first_url).query)["token"][0]
|
|
email_projection = status_access.get_public_projection(
|
|
self.session,
|
|
tracking_id=email_summary["tracking_id"],
|
|
token=first_token,
|
|
observed_at=NOW + timedelta(minutes=9),
|
|
)
|
|
self.assertEqual(email_submission.receipt_id, email_projection["receipt_id"])
|
|
self.assertTrue(
|
|
status_access.request_email_link(
|
|
self.session,
|
|
tracking_id=email_summary["tracking_id"],
|
|
email="ada@example.test",
|
|
requested_at=NOW + timedelta(minutes=10),
|
|
)
|
|
)
|
|
with self.assertRaises(FormStatusUnavailable):
|
|
status_access.get_public_projection(
|
|
self.session,
|
|
tracking_id=email_summary["tracking_id"],
|
|
token=first_token,
|
|
observed_at=NOW + timedelta(minutes=11),
|
|
)
|
|
second_url = notifications.requests[-1].action_url
|
|
second_token = parse_qs(urlsplit(second_url).query)["token"][0]
|
|
self.assertNotEqual(first_token, second_token)
|
|
|
|
policy = status_access.upsert_policy(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="permanent_link",
|
|
enabled=True,
|
|
email_field_key=None,
|
|
expected_revision=policy.revision,
|
|
recorded_at=NOW + timedelta(minutes=12),
|
|
)
|
|
permanent_draft = runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada", "delivery": "portal"},
|
|
idempotency_key="status-permanent-create",
|
|
recorded_at=NOW + timedelta(minutes=13),
|
|
)
|
|
permanent_submission = runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=permanent_draft.instance_id,
|
|
expected_revision=1,
|
|
values=permanent_draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(),
|
|
idempotency_key="status-permanent-submit",
|
|
recorded_at=NOW + timedelta(minutes=14),
|
|
)
|
|
permanent_summary = status_access.access_summary_for_instance(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
instance_id=permanent_submission.instance_id,
|
|
)
|
|
permanent_projection = status_access.get_public_projection(
|
|
self.session,
|
|
tracking_id=permanent_summary["tracking_id"],
|
|
token=None,
|
|
observed_at=NOW + timedelta(minutes=15),
|
|
)
|
|
self.assertEqual("submitted", permanent_projection["status"])
|
|
policy = status_access.upsert_policy(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
mode="permanent_link",
|
|
enabled=False,
|
|
email_field_key=None,
|
|
expected_revision=policy.revision,
|
|
recorded_at=NOW + timedelta(minutes=16),
|
|
)
|
|
self.assertFalse(policy.enabled)
|
|
with self.assertRaises(FormStatusUnavailable):
|
|
status_access.get_public_projection(
|
|
self.session,
|
|
tracking_id=permanent_summary["tracking_id"],
|
|
token=None,
|
|
observed_at=NOW + timedelta(minutes=17),
|
|
)
|
|
|
|
def test_native_acknowledgement_is_bound_to_actor_and_exact_payload(self) -> None:
|
|
draft = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="ack-create",
|
|
recorded_at=NOW,
|
|
)
|
|
signature = self.runtime.acknowledge(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
statement_id="submission-correct",
|
|
statement_version="1",
|
|
values=draft.values,
|
|
attachment_refs=(),
|
|
accepted_at=NOW + timedelta(minutes=1),
|
|
idempotency_key="ack-1",
|
|
)
|
|
submitted = self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=draft.instance_id,
|
|
expected_revision=1,
|
|
values=draft.values,
|
|
attachment_refs=(),
|
|
signature_refs=(signature,),
|
|
idempotency_key="ack-submit",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
self.assertEqual(
|
|
"accepted", submitted.metadata["evidence_verification"][0]["state"]
|
|
)
|
|
|
|
second = self.runtime.create_instance(
|
|
self.session,
|
|
self.principal,
|
|
definition_ref=self.definition.reference,
|
|
values={"name": "Ada"},
|
|
idempotency_key="ack-create-2",
|
|
recorded_at=NOW,
|
|
)
|
|
stale_signature = self.runtime.acknowledge(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=second.instance_id,
|
|
expected_revision=1,
|
|
statement_id="submission-correct",
|
|
statement_version="1",
|
|
values=second.values,
|
|
attachment_refs=(),
|
|
accepted_at=NOW + timedelta(minutes=1),
|
|
idempotency_key="ack-2",
|
|
)
|
|
with self.assertRaisesRegex(FormRuntimeError, "failed final verification"):
|
|
self.runtime.submit_instance(
|
|
self.session,
|
|
self.principal,
|
|
instance_id=second.instance_id,
|
|
expected_revision=1,
|
|
values={"name": "Changed after acknowledgement"},
|
|
attachment_refs=(),
|
|
signature_refs=(stale_signature,),
|
|
idempotency_key="ack-stale-submit",
|
|
recorded_at=NOW + timedelta(minutes=2),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|