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.institutional import ( CAPABILITY_FORM_DEFINITIONS, FormDefinition, FormFieldDefinition, InstitutionalReference, ServiceBinding, ServiceDefinition, ServiceLaunchRequest, TemporalRevision, ) 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 ( FormInstanceEvent, FormInstanceIdentity, FormInstanceRevision, ) from govoplan_forms_runtime.backend.service import ( FormRuntimeError, FormRuntimeService, FormsServiceLauncher, ) 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) -> None: self.provider = provider def has_capability(self, name: str) -> bool: return name == CAPABILITY_FORM_DEFINITIONS def require_capability(self, name: str) -> object: if name != CAPABILITY_FORM_DEFINITIONS: raise KeyError(name) return self.provider 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"), ), ), 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__, ): 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", ) if __name__ == "__main__": unittest.main()