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

feat: add configurable application status access

Browse Source
This commit is contained in:
zemion
2026-08-19 12:33:34 +02:00
parent d83bb92ec8
commit 87d12519b7
17 changed files with 2073 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_forms_runtime.py
+285
View File
@@ -3,6 +3,7 @@ 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
@@ -20,6 +21,7 @@ from govoplan_core.core.institutional import (
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,
@@ -34,6 +36,9 @@ from govoplan_forms_runtime.backend.db.models import (
FormInstanceRevision,
FormIntakeProfile,
FormIntakeSession,
FormStatusAccessGrant,
FormStatusAccessPolicy,
FormStatusAccessToken,
)
from govoplan_forms_runtime.backend.service import (
FormRuntimeError,
@@ -42,6 +47,11 @@ from govoplan_forms_runtime.backend.service import (
)
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)
@@ -66,6 +76,53 @@ class Registry:
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(
*,
@@ -101,6 +158,12 @@ def form_definition(
value_type="choice",
options=("portal", "mail"),
),
FormFieldDefinition(
key="email",
label="Email",
value_type="email",
constraints={"max_length": 320},
),
),
publication_state="published",
allow_drafts=True,
@@ -121,6 +184,9 @@ class FormsRuntimeTests(unittest.TestCase):
FormIntakeProfile.__table__,
FormIntakeSession.__table__,
FormAssistedConfirmation.__table__,
FormStatusAccessPolicy.__table__,
FormStatusAccessGrant.__table__,
FormStatusAccessToken.__table__,
FormAcknowledgement.__table__,
):
table.create(self.engine)
@@ -987,6 +1053,225 @@ class FormsRuntimeTests(unittest.TestCase):
),
)
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,
tests/test_migrations.py
+4 -1
View File
@@ -35,12 +35,15 @@ class FormsRuntimeMigrationTests(unittest.TestCase):
"form_intake_profiles",
"form_intake_sessions",
"form_assisted_confirmations",
"form_status_access_policies",
"form_status_access_grants",
"form_status_access_tokens",
"form_acknowledgements",
}.issubset(inspect(engine).get_table_names())
)
with engine.connect() as connection:
self.assertIn(
"c5f7a9b1d3e4",
"d6a8b0c2e4f6",
set(MigrationContext.configure(connection).get_current_heads()),
)
finally: