[#4267] Add management command to migrate existing configurations

src/openforms/contrib/zgw/clients/catalogi.py

@@ -1,4 +1,5 @@
-from typing import Callable
+from typing import Any, Callable
+from uuid import UUID
 
 from zgw_consumers.nlx import NLXClient
 
@@ -38,6 +39,11 @@ def get_all_informatieobjecttypen(self) -> list[dict]:
         all_data = pagination_helper(self, data)
         return list(all_data)
 
+    def get_informatieobjecttype(self, uuid: str | UUID) -> dict[str, Any]:
+        response = self.get(f"informatieobjecttypen/{uuid}")
+        response.raise_for_status()
+        return response.json()
+
     def list_statustypen(self, zaaktype: str) -> list[dict]:
         query = {"zaaktype": zaaktype}
 

src/openforms/registrations/contrib/objects_api/management/commands/migrate_objects_api_config.py

@@ -0,0 +1,113 @@
+from typing import Any
+
+from django.core.management import BaseCommand
+
+import requests
+
+from openforms.forms.models import FormRegistrationBackend
+
+from ...client import get_catalogi_client
+from ...plugin import PLUGIN_IDENTIFIER as OBJECTS_API_PLUGIN_IDENTIFIER
+
+IOT_FIELDS = (
+    "informatieobjecttype_submission_report",
+    "informatieobjecttype_submission_csv",
+    "informatieobjecttype_attachment",
+)
+
+
+class SkipBackend(Exception):
+    """The backend options were already migrated."""
+
+
+class InvalidBackend(Exception):
+    """The backend options are invalid and can't be migrated."""
+
+    def __init__(self, message: str, /) -> None:
+        self.message = message
+
+
+def migrate_registration_backend(options: dict[str, Any]) -> None:
+    # idempotent check:
+    if "catalogus_domein" in options:
+        raise SkipBackend
+
+    iot_omschrijving: dict[str, str] = {}
+
+    with get_catalogi_client(options["objects_api_group"]) as catalogi_client:
+        catalogus_domein: str | None = None
+        catalogus_rsin: str | None = None
+
+        for field in IOT_FIELDS:
+            if url := options.get(field):
+                try:
+                    uuid = url.rsplit("/", 1)[1]
+                except IndexError:
+                    raise InvalidBackend(f"{field}: {url} is invalid")
+
+                try:
+                    iot = catalogi_client.get_informatieobjecttype(uuid)
+                except requests.HTTPError as e:
+                    if (status := e.response.status_code) == 404:
+                        raise InvalidBackend(f"{field}: {url} does not exist")
+                    else:
+                        raise RuntimeError(
+                            f"Encountered {status} error when fetching informatieobjecttype {uuid}"
+                        )
+
+                catalogus_resp = catalogi_client.get(iot["catalogus"])
+                if not catalogus_resp.ok:
+                    raise RuntimeError(
+                        f"Encountered {catalogus_resp.status_code} error when fetching catalog {iot['catalogus']}"
+                    )
+                catalogus = catalogus_resp.json()
+
+                if catalogus_domein is None and catalogus_rsin is None:
+                    catalogus_domein = catalogus["domein"]
+                    catalogus_rsin = catalogus["rsin"]
+                else:
+                    if (
+                        catalogus["domein"] != catalogus_domein
+                        or catalogus["rsin"] != catalogus_rsin
+                    ):
+                        raise InvalidBackend(
+                            "Informatieobjecttypes don't share the same catalog"
+                        )
+
+                iot_omschrijving[field] = iot["omschrijving"]
+
+    if catalogus_domein is None and catalogus_rsin is None:
+        options["catalogus_domein"] = catalogus_domein
+        options["catalogus_rsin"] = catalogus_rsin
+
+        for field in IOT_FIELDS:
+            if field in iot_omschrijving:
+                options[field] = iot_omschrijving[field]
+
+
+class Command(BaseCommand):
+    help = (
+        "Migrate existing Objects API form registration backends "
+        "to switch from an informatieobjectype URL to a omschrijving and catalogus."
+    )
+
+    def handle(self, **options):
+        for registration_backend in FormRegistrationBackend.objects.filter(
+            backend=OBJECTS_API_PLUGIN_IDENTIFIER
+        ):
+            backend_name = registration_backend.name
+
+            try:
+                migrate_registration_backend(registration_backend.options)
+            except SkipBackend:
+                self.stdout.write(f"{backend_name!r} was already migrated")
+            except InvalidBackend as e:
+                self.stderr.write(
+                    f"{backend_name!r} configuration is invalid: {e.message!r}"
+                )
+            except RuntimeError as e:
+                self.stderr.write(
+                    f"Encountered runtime error when trying to migrate {backend_name!r}: {e.args[0]!r}"
+                )
+            else:
+                registration_backend.save()

src/openforms/registrations/contrib/objects_api/tests/test_management_commands.py

@@ -0,0 +1,93 @@
+from pathlib import Path
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from openforms.forms.models import FormRegistrationBackend
+from openforms.forms.tests.factories import FormRegistrationBackendFactory
+from openforms.utils.tests.vcr import OFVCRMixin
+
+from ..management.commands.migrate_objects_api_config import (
+    InvalidBackend,
+    SkipBackend,
+    migrate_registration_backend,
+)
+from ..plugin import PLUGIN_IDENTIFIER
+from .factories import ObjectsAPIGroupConfigFactory
+
+
+class MigrateObjectsAPIV2ConfigTests(OFVCRMixin, TestCase):
+
+    VCR_TEST_FILES = Path(__file__).parent / "files"
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+
+        cls.objects_api_group = ObjectsAPIGroupConfigFactory.create(
+            for_test_docker_compose=True
+        )
+
+    def test_already_migrated(self):
+        with self.assertRaises(SkipBackend):
+            migrate_registration_backend(
+                {
+                    "catalogus_domein": "TEST DOMEIN",
+                    "catalogus_rsin": "000000000",
+                }
+            )
+
+    def test_no_iot_fields(self):
+        options = {}
+        migrate_registration_backend(options)
+
+        self.assertEqual(options, {})
+
+    def test_invalid_iot(self):
+        options = {
+            # This should be an URL:
+            "informatieobjecttype_attachment": "--wrong--"
+        }
+        with self.assertRaises(InvalidBackend):
+            migrate_registration_backend(options)
+
+    def test_unknown_iot(self):
+        options = {
+            # 996946c4-92e8-4bf2-a18f-3ef41dbb423f does not exist on the Docker Compose instance:
+            "informatieobjecttype_attachment": "http://localhost:8003/catalogi/api/v1/informatieobjecttypen/996946c4-92e8-4bf2-a18f-3ef41dbb423f"
+        }
+        with self.assertRaises(InvalidBackend):
+            migrate_registration_backend(options)
+
+    def test_iots_different_catalogs(self):
+        # TODO update fixtures first
+        options = {
+            "informatieobjecttype_attachment": "",
+            "informatieobjecttype_submission_csv": "",
+        }
+        with self.assertRaises(InvalidBackend):
+            migrate_registration_backend(options)
+
+    def test_valid_config(self):
+        # TODO update fixtures first
+        backend: FormRegistrationBackend = FormRegistrationBackendFactory.create(
+            backend=PLUGIN_IDENTIFIER,
+            options={
+                "informatieobjecttype_attachment": "",
+                "informatieobjecttype_submission_csv": "",
+            },
+        )
+
+        call_command("migrate_objects_api_config")
+
+        backend.refresh_from_db()
+
+        self.assertEqual(
+            backend.options,
+            {
+                "informatieobjecttype_attachment": "",  # TODO oms
+                "informatieobjecttype_submission_csv": "",  # TODO oms
+                "catalogus_domein": "TEST",
+                "catalogus_rsin": "000000000",
+            },
+        )