GovOPlaN/govoplan-forms-runtime
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add governed public form intake

Browse Source
This commit is contained in:
zemion
2026-08-06 12:42:20 +02:00
parent d4bbdd079f
commit 3c292b8c07
25 changed files with 4715 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_forms_runtime.py
+166
View File
@@ -26,10 +26,13 @@ from govoplan_forms.backend.service import (
record_form_definition,
)
from govoplan_forms_runtime.backend.db.models import (
FormAcknowledgement,
FormHandoffEffect,
FormInstanceEvent,
FormInstanceIdentity,
FormInstanceRevision,
FormIntakeProfile,
FormIntakeSession,
)
from govoplan_forms_runtime.backend.service import (
FormRuntimeError,
@@ -37,6 +40,7 @@ from govoplan_forms_runtime.backend.service import (
FormsServiceLauncher,
)
from govoplan_forms_runtime.backend.handoffs import FormHandoffService
from govoplan_forms_runtime.backend.intake import FormIntakeError, FormIntakeService
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
@@ -113,6 +117,9 @@ class FormsRuntimeTests(unittest.TestCase):
FormInstanceRevision.__table__,
FormInstanceEvent.__table__,
FormHandoffEffect.__table__,
FormIntakeProfile.__table__,
FormIntakeSession.__table__,
FormAcknowledgement.__table__,
):
table.create(self.engine)
self.session = Session(self.engine)
@@ -683,6 +690,165 @@ class FormsRuntimeTests(unittest.TestCase):
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_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()