govoplan-forms/tests/test_forms.py

from __future__ import annotations

from dataclasses import dataclass, replace
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,
    FormConditionExpression,
    FormFieldDefinition,
    FormLocalization,
    FormPageDefinition,
    FormSectionDefinition,
    InstitutionalReference,
    TemporalRevision,
)
from govoplan_forms.backend.db.models import FormDefinitionRevision
from govoplan_forms.backend.service import (
    FormDefinitionStoreError,
    SqlFormDefinitionProvider,
    assess_form_definition_fragment,
    export_form_definition_fragment,
    form_definition_diagnostics,
    form_definition_history,
    import_form_definition_fragment,
    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",
            )

    def test_pages_conditions_and_localization_are_validated(self) -> None:
        item = definition()
        composed = replace(
            item,
            fields=(
                item.fields[0],
                replace(
                    item.fields[1],
                    visibility_condition=FormConditionExpression(
                        kind="predicate",
                        field_key="name",
                        operator="is_not_empty",
                    ),
                ),
            ),
            pages=(
                FormPageDefinition(
                    key="application",
                    title="Application",
                    sections=(
                        FormSectionDefinition(
                            key="details",
                            title="Details",
                            field_keys=("name", "delivery"),
                        ),
                    ),
                ),
            ),
            fallback_locale="de",
            localizations=(
                FormLocalization(
                    locale="de",
                    title="Antrag",
                    field_labels={"name": "Name", "delivery": "Zustellung"},
                    option_labels={"delivery": {"portal": "Portal", "mail": "Post"}},
                    page_titles={"application": "Antrag"},
                    section_titles={"details": "Angaben"},
                ),
            ),
        )
        stored = record_form_definition(
            self.session,
            self.principal,
            definition=composed,
        )
        self.assertEqual("application", stored.pages[0].key)
        self.assertEqual((), form_definition_diagnostics(stored))

        cyclic = replace(
            definition(revision="2"),
            fields=(
                replace(
                    item.fields[0],
                    visibility_condition=FormConditionExpression(
                        kind="predicate",
                        field_key="delivery",
                        operator="eq",
                        value="portal",
                    ),
                ),
                replace(
                    item.fields[1],
                    visibility_condition=FormConditionExpression(
                        kind="predicate",
                        field_key="name",
                        operator="is_not_empty",
                    ),
                ),
            ),
        )
        with self.assertRaisesRegex(FormDefinitionStoreError, "dependency cycle"):
            record_form_definition(
                self.session,
                self.principal,
                definition=cyclic,
                expected_revision="1",
            )

    def test_package_fragment_is_verified_assessed_and_imported_as_draft(self) -> None:
        source = definition(tenant_id="tenant-source")
        fragment = export_form_definition_fragment(
            source,
            exported_at=NOW,
            exported_by="source-account",
        )
        assessment = assess_form_definition_fragment(
            self.session,
            self.principal,
            fragment=fragment,
        )
        self.assertEqual("create", assessment["outcome"])
        self.assertTrue(assessment["requires_remap"])

        imported = import_form_definition_fragment(
            self.session,
            self.principal,
            fragment=fragment,
            target_form_id="local-permit",
            target_key="local-permit",
            expected_revision=None,
            change_reason="Import reviewed package.",
            recorded_at=NOW,
        )
        self.assertEqual("tenant-1", imported.reference.tenant_id)
        self.assertEqual("draft", imported.publication_state)
        self.assertEqual(
            "tenant-source",
            imported.metadata["package_import"]["source_tenant_id"],
        )

        tampered = dict(fragment)
        tampered["definition"] = {**fragment["definition"], "title": "Tampered"}
        with self.assertRaisesRegex(FormDefinitionStoreError, "digest"):
            assess_form_definition_fragment(
                self.session,
                self.principal,
                fragment=tampered,
            )


if __name__ == "__main__":
    unittest.main()
    (assess_form_definition_fragment,)
    (export_form_definition_fragment,)
    (form_definition_diagnostics,)
    (import_form_definition_fragment,)