Compose v0.1.35 resident permit configuration package

tests/test_configuration_package_artifacts.py

@@ -5,9 +5,17 @@ import json
 from pathlib import Path
 import unittest
 
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
 from govoplan_core.core.configuration_packages import (
+    ConfigurationApplyResult,
+    ConfigurationExportResult,
     ConfigurationPackageManifest,
+    ConfigurationPlanItem,
     ConfigurationPreflightContext,
+    ConfigurationPreflightResult,
+    ConfigurationProviderDescription,
     configuration_package_claim_issues,
     dry_run_configuration_package,
 )
@@ -54,10 +62,19 @@ class ConfigurationPackageArtifactTests(unittest.TestCase):
                     f"Missing repository for {requirement.module_id}",
                 )
 
+            provider_module_ids = tuple(
+                sorted({fragment.module_id for fragment in manifest.fragments})
+            )
+            providers = tuple(_ArtifactProvider(module_id) for module_id in provider_module_ids)
+            supplied_data = {
+                str(item["key"]): _sample_value(item)
+                for item in manifest.data_requirements
+            }
             result = dry_run_configuration_package(
                 manifest,
-                (),
+                providers,
                 ConfigurationPreflightContext(
+                    supplied_data=supplied_data,
                     installed_modules={
                         item.module_id: item.version or "workspace"
                         for item in manifest.required_modules
@@ -74,6 +91,155 @@ class ConfigurationPackageArtifactTests(unittest.TestCase):
         self.assertIn("product.governed-data-assurance", package_ids)
         self.assertIn("product.service-to-decision", package_ids)
 
+    def test_service_to_decision_package_imports_its_form_as_an_idempotent_local_draft(self) -> None:
+        from govoplan_forms.backend.configuration_provider import (
+            _apply_definition,
+            _preflight_definition,
+        )
+        from govoplan_forms.backend.db.models import FormDefinitionRevision
+        from govoplan_forms.backend.service import get_form_definition
+
+        package = ConfigurationPackageManifest.from_mapping(json.loads(
+            (PACKAGE_ROOT / "product/service-to-decision/package.json").read_text(
+                encoding="utf-8"
+            )
+        ))
+        fragment = next(
+            item
+            for item in package.fragments
+            if item.module_id == "forms" and item.fragment_type == "definition"
+        )
+        context = ConfigurationPreflightContext(
+            tenant_id="tenant-reference-test",
+            operator_user_id="operator-1",
+            operator_scopes=frozenset({"system:governance:write"}),
+        )
+        engine = create_engine("sqlite+pysqlite:///:memory:")
+        FormDefinitionRevision.__table__.create(engine)
+        session = Session(engine)
+        try:
+            preflight = _preflight_definition(session, fragment, context)
+            applied = _apply_definition(session, fragment, context)
+            session.commit()
+            replay = _apply_definition(session, fragment, context)
+            imported = get_form_definition(
+                session,
+                type("Principal", (), {"tenant_id": "tenant-reference-test"})(),
+                form_id="resident-parking-permit-application",
+            )
+        finally:
+            session.close()
+            engine.dispose()
+
+        self.assertFalse(any(item.severity == "blocker" for item in preflight.diagnostics))
+        self.assertEqual("create", preflight.plan[0].action)
+        self.assertEqual(1, len(applied.created_refs))
+        self.assertEqual({}, replay.created_refs)
+        self.assertIsNotNone(imported)
+        assert imported is not None
+        self.assertEqual("tenant-reference-test", imported.reference.tenant_id)
+        self.assertEqual("draft", imported.publication_state)
+        self.assertEqual("de", imported.fallback_locale)
+
+    def test_service_to_decision_package_materializes_its_tenant_workflow_idempotently(self) -> None:
+        from govoplan_core.core.configuration_packages import _resolve_fragment_data_references
+        from govoplan_workflow_engine.backend.configuration_provider import (
+            apply_workflow_definitions,
+            preflight_workflow_definitions,
+        )
+        from govoplan_workflow_engine.backend.db.models import (
+            WorkflowDefinition,
+            WorkflowDefinitionRevision,
+        )
+
+        package = ConfigurationPackageManifest.from_mapping(json.loads(
+            (PACKAGE_ROOT / "product/service-to-decision/package.json").read_text(
+                encoding="utf-8"
+            )
+        ))
+        fragment = next(
+            item
+            for item in package.fragments
+            if item.module_id == "workflow_engine"
+        )
+        context = ConfigurationPreflightContext(
+            tenant_id="tenant-reference-test",
+            supplied_data={
+                "responsible_group_slug": "traffic-permits",
+                "responsible_group_name": "Traffic permits",
+            },
+        )
+        resolved = _resolve_fragment_data_references(
+            fragment,
+            context.supplied_data,
+        )
+        engine = create_engine("sqlite+pysqlite:///:memory:")
+        WorkflowDefinition.__table__.create(engine)
+        WorkflowDefinitionRevision.__table__.create(engine)
+        session = Session(engine)
+        try:
+            preflight = preflight_workflow_definitions(session, resolved, context)
+            applied = apply_workflow_definitions(
+                session,
+                resolved,
+                context,
+                registry=None,
+            )
+            replay = apply_workflow_definitions(
+                session,
+                resolved,
+                context,
+                registry=None,
+            )
+            session.commit()
+        finally:
+            session.close()
+            engine.dispose()
+
+        self.assertFalse(any(item.severity == "blocker" for item in preflight.diagnostics))
+        self.assertEqual("create", preflight.plan[0].action)
+        self.assertIn("resident-parking-permit-review", applied.created_refs)
+        self.assertEqual({}, replay.created_refs)
+        self.assertEqual({}, replay.updated_refs)
+
+
+class _ArtifactProvider:
+    def __init__(self, module_id: str) -> None:
+        self.module_id = module_id
+
+    def describe(self) -> ConfigurationProviderDescription:
+        return ConfigurationProviderDescription(module_id=self.module_id)
+
+    def preflight(self, fragment, context) -> ConfigurationPreflightResult:
+        del context
+        return ConfigurationPreflightResult(plan=(ConfigurationPlanItem(
+            action="create",
+            module_id=fragment.module_id,
+            fragment_type=fragment.fragment_type,
+            fragment_id=fragment.fragment_id,
+        ),))
+
+    def apply(self, fragment, supplied_data, context) -> ConfigurationApplyResult:
+        del fragment, supplied_data, context
+        return ConfigurationApplyResult()
+
+    def export(self, selection, context) -> ConfigurationExportResult:
+        del selection, context
+        return ConfigurationExportResult()
+
+    def health(self, import_result, context):
+        del import_result, context
+        return ()
+
+
+def _sample_value(requirement: dict[str, object]) -> object:
+    data_type = str(requirement.get("data_type") or requirement.get("type") or "string")
+    if data_type == "boolean":
+        return False
+    if data_type in {"integer", "number"}:
+        return 1
+    return f"fixture-{requirement['key']}"
+
 
 if __name__ == "__main__":
     unittest.main()