govoplan-forms/tests/test_configuration_provider.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime
import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from govoplan_core.core.configuration_packages import (
    ConfigurationPackageFragment,
    ConfigurationPreflightContext,
    ConfigurationProvider,
)
from govoplan_core.core.institutional import (
    FormDefinition,
    FormFieldDefinition,
    FormLocalization,
    FormPageDefinition,
    FormSectionDefinition,
    InstitutionalReference,
    TemporalRevision,
)
from govoplan_forms.backend.configuration_provider import (
    FORMS_CONFIGURATION_CAPABILITY,
    SqlFormsConfigurationProvider,
    _apply_definition,
    _preflight_definition,
)
from govoplan_forms.backend.db.models import FormDefinitionRevision
from govoplan_forms.backend.manifest import get_manifest
from govoplan_forms.backend.service import (
    export_form_definition_fragment,
    get_form_definition,
)


@dataclass(frozen=True)
class Principal:
    tenant_id: str
    account_id: str = "operator-1"


def source_definition(title: str = "Resident parking permit") -> FormDefinition:
    return FormDefinition(
        reference=InstitutionalReference(
            kind="form",
            owner_module="forms",
            object_id="resident-parking-permit-application",
            tenant_id="reference-package",
            version="3",
        ),
        key="resident-parking-permit-application",
        temporal=TemporalRevision(
            revision="3",
            recorded_at=datetime(2026, 8, 22, tzinfo=UTC),
            change_reason="Reference package revision.",
        ),
        title=title,
        description="Apply for a resident parking permit through a digital or assisted channel.",
        fields=(
            FormFieldDefinition(key="applicant_name", label="Name", required=True, constraints={"min_length": 2, "max_length": 200}),
            FormFieldDefinition(key="applicant_email", label="Email", required=True, constraints={"format": "email"}),
            FormFieldDefinition(key="residence_address", label="Primary residence", required=True, constraints={"max_length": 500}),
            FormFieldDefinition(key="licence_plate", label="Licence plate", required=True, constraints={"max_length": 20}),
        ),
        publication_state="published",
        max_attachments=4,
        policy_refs=("law:resident-parking-permit", "records:resident-parking-permit"),
        handoff_kinds=("case", "workflow"),
        pages=(
            FormPageDefinition(
                key="application",
                title="Application",
                sections=(
                    FormSectionDefinition(
                        key="applicant-and-vehicle",
                        title="Applicant and vehicle",
                        field_keys=("applicant_name", "applicant_email", "residence_address", "licence_plate"),
                    ),
                ),
            ),
        ),
        fallback_locale="de",
        localizations=(
            FormLocalization(
                locale="de",
                title="Anwohnerparkausweis beantragen",
                description="Einen Anwohnerparkausweis digital oder mit Unterstützung beantragen.",
                field_labels={
                    "applicant_name": "Name",
                    "applicant_email": "E-Mail-Adresse",
                    "residence_address": "Hauptwohnsitz",
                    "licence_plate": "Kennzeichen",
                },
                page_titles={"application": "Antrag"},
                section_titles={"applicant-and-vehicle": "Antragstellende Person und Fahrzeug"},
            ),
        ),
    )


class FormsConfigurationProviderTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        FormDefinitionRevision.__table__.create(self.engine)
        self.session = Session(self.engine)
        self.context = ConfigurationPreflightContext(
            tenant_id="tenant-1",
            operator_user_id="operator-1",
            operator_scopes=frozenset({"system:governance:write"}),
            installed_modules={"forms": "0.1.20"},
            capabilities=frozenset({FORMS_CONFIGURATION_CAPABILITY}),
        )

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    def fragment(self, definition: FormDefinition | None = None, *, on_conflict: str = "new_revision") -> ConfigurationPackageFragment:
        return ConfigurationPackageFragment(
            module_id="forms",
            fragment_type="definition",
            fragment_id="resident-parking-permit-application",
            payload={
                "fragment": export_form_definition_fragment(
                    definition or source_definition(),
                    exported_at=datetime(2026, 8, 22, 12, tzinfo=UTC),
                    exported_by="package-author",
                ),
                "on_conflict": on_conflict,
                "change_reason": "Install the reviewed resident parking permit reference form.",
            },
        )

    def test_provider_is_registered_and_runtime_checkable(self) -> None:
        provider = get_manifest().capability_factories[FORMS_CONFIGURATION_CAPABILITY](None)  # type: ignore[arg-type]

        self.assertIsInstance(provider, ConfigurationProvider)
        self.assertIsInstance(provider, SqlFormsConfigurationProvider)
        self.assertEqual(("definition",), provider.describe().fragment_types)

    def test_import_is_tenant_local_draft_and_same_source_replay_is_noop(self) -> None:
        fragment = self.fragment()

        preflight = _preflight_definition(self.session, fragment, self.context)
        applied = _apply_definition(self.session, fragment, self.context)
        self.session.commit()
        replay_preflight = _preflight_definition(self.session, fragment, self.context)
        replay = _apply_definition(self.session, fragment, self.context)

        self.assertEqual("create", preflight.plan[0].action)
        self.assertEqual(1, len(applied.created_refs))
        imported = get_form_definition(
            self.session,
            Principal("tenant-1"),
            form_id="resident-parking-permit-application",
        )
        assert imported is not None
        self.assertEqual("tenant-1", imported.reference.tenant_id)
        self.assertEqual("draft", imported.publication_state)
        self.assertEqual(
            "reference-package",
            imported.metadata["package_import"]["source_tenant_id"],
        )
        self.assertEqual("noop", replay_preflight.plan[0].action)
        self.assertEqual({}, replay.created_refs)
        self.assertEqual({}, replay.updated_refs)

    def test_preserve_reports_conflict_and_missing_authority_blocks(self) -> None:
        first = self.fragment()
        _apply_definition(self.session, first, self.context)
        self.session.commit()
        changed = self.fragment(source_definition("Changed reference"), on_conflict="preserve")
        unauthorized = ConfigurationPreflightContext(
            tenant_id="tenant-1",
            operator_user_id="operator-2",
        )

        conflict = _preflight_definition(self.session, changed, self.context)
        denied = _preflight_definition(self.session, changed, unauthorized)

        self.assertIn(
            "forms_configuration_conflict",
            {item.code for item in conflict.diagnostics},
        )
        self.assertIn(
            "forms_configuration_write_scope_required",
            {item.code for item in denied.diagnostics},
        )


if __name__ == "__main__":
    unittest.main()