feat: implement immutable form definitions

tests/test_forms.py

@@ -0,0 +1,167 @@
+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 (
+    FormDefinition,
+    FormFieldDefinition,
+    InstitutionalReference,
+    TemporalRevision,
+)
+from govoplan_forms.backend.db.models import FormDefinitionRevision
+from govoplan_forms.backend.service import (
+    FormDefinitionStoreError,
+    SqlFormDefinitionProvider,
+    form_definition_history,
+    record_form_definition,
+)
+
+
+NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC)
+
+
+@dataclass
+class Principal:
+    tenant_id: str = "tenant-1"
+    account_id: str = "account-1"
+
+
+def definition(
+    *,
+    revision: str = "1",
+    state: str = "published",
+    tenant_id: str = "tenant-1",
+) -> FormDefinition:
+    recorded_at = NOW + timedelta(minutes=int(revision) - 1)
+    return FormDefinition(
+        reference=InstitutionalReference(
+            kind="form",
+            owner_module="forms",
+            object_id="permit-application",
+            tenant_id=tenant_id,
+            version=revision,
+        ),
+        key="permit-application",
+        temporal=TemporalRevision(
+            revision=revision,
+            recorded_at=recorded_at,
+            change_reason="Initial schema." if revision == "1" else "Revise schema.",
+        ),
+        title="Permit application",
+        fields=(
+            FormFieldDefinition(
+                key="name",
+                label="Name",
+                required=True,
+                constraints={"min_length": 2, "max_length": 200},
+            ),
+            FormFieldDefinition(
+                key="delivery",
+                label="Delivery",
+                value_type="choice",
+                options=("portal", "mail"),
+            ),
+        ),
+        publication_state=state,  # type: ignore[arg-type]
+        allow_drafts=True,
+        max_attachments=2,
+        handoff_kinds=("case",),
+    )
+
+
+class FormsTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        FormDefinitionRevision.__table__.create(self.engine)
+        self.session = Session(self.engine)
+        self.principal = Principal()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_provider_returns_exact_published_revision_and_history(self) -> None:
+        first = record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(),
+        )
+        provider = SqlFormDefinitionProvider()
+
+        self.assertEqual(
+            (first,),
+            tuple(
+                provider.list_form_definitions(
+                    self.session,
+                    self.principal,
+                    tenant_id="tenant-1",
+                )
+            ),
+        )
+        exact = provider.get_form_definition(
+            self.session,
+            self.principal,
+            reference=first.reference,
+            effective_at=NOW,
+        )
+        self.assertEqual("1", exact.temporal.revision if exact else None)
+
+        record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(revision="2"),
+            expected_revision="1",
+        )
+        self.assertEqual(
+            ["2", "1"],
+            [
+                item.temporal.revision
+                for item in form_definition_history(
+                    self.session,
+                    self.principal,
+                    form_id="permit-application",
+                )
+            ],
+        )
+
+    def test_replay_occ_and_tenant_boundaries_fail_closed(self) -> None:
+        first = record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(),
+        )
+        replay = record_form_definition(
+            self.session,
+            self.principal,
+            definition=definition(),
+        )
+        self.assertEqual(first, replay)
+
+        with self.assertRaisesRegex(FormDefinitionStoreError, "stale"):
+            record_form_definition(
+                self.session,
+                self.principal,
+                definition=definition(revision="2"),
+                expected_revision="0",
+            )
+        with self.assertRaisesRegex(FormDefinitionStoreError, "cross tenants"):
+            record_form_definition(
+                self.session,
+                self.principal,
+                definition=definition(tenant_id="tenant-2"),
+            )
+        with self.assertRaisesRegex(Exception, "cross tenants"):
+            SqlFormDefinitionProvider().list_form_definitions(
+                self.session,
+                Principal("tenant-2"),
+                tenant_id="tenant-1",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_migrations.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+import unittest
+
+from alembic.runtime.migration import MigrationContext
+from sqlalchemy import create_engine, inspect
+
+from govoplan_core.db.migrations import migrate_database
+from govoplan_forms.backend.manifest import get_manifest
+
+
+class FormsMigrationTests(unittest.TestCase):
+    def test_fresh_migration_creates_definition_store_and_head(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-forms-migration-") as directory:
+            url = f"sqlite:///{Path(directory) / 'forms.db'}"
+            migrate_database(
+                database_url=url,
+                enabled_modules=("forms",),
+                manifest_factories=(get_manifest,),
+            )
+            engine = create_engine(url)
+            try:
+                self.assertIn(
+                    "form_definition_revisions",
+                    inspect(engine).get_table_names(),
+                )
+                with engine.connect() as connection:
+                    self.assertIn(
+                        "e1f2a3b4c5d6",
+                        set(MigrationContext.configure(connection).get_current_heads()),
+                    )
+            finally:
+                engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()