expose the ZGW import/export functionality through the admin page

Author: swrichards

src/open_inwoner/openzaak/admin.py

@@ -1,14 +1,27 @@
+import datetime
+import logging
+
 from django.contrib import admin, messages
 from django.core.exceptions import ValidationError
 from django.db.models import BooleanField, Count, ExpressionWrapper, Q
-from django.forms import ModelForm, Textarea
+from django.forms import Form, ModelForm, Textarea
 from django.forms.models import BaseInlineFormSet
+from django.http import HttpResponseRedirect, StreamingHttpResponse
+from django.template.defaultfilters import filesizeformat
+from django.template.response import TemplateResponse
+from django.urls import path, reverse
 from django.utils.translation import gettext_lazy as _, ngettext
 
 from import_export.admin import ImportExportMixin
+from privates.storages import PrivateMediaFileSystemStorage
 from solo.admin import SingletonModelAdmin
 
 from open_inwoner.ckeditor5.widgets import CKEditorWidget
+from open_inwoner.openzaak.import_export import (
+    CatalogusConfigExport,
+    CatalogusConfigImport,
+)
+from open_inwoner.utils.forms import LimitedUploadFileField
 
 from .models import (
     CatalogusConfig,
@@ -22,6 +35,8 @@
     ZGWApiGroupConfig,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class ZGWApiGroupConfig(admin.StackedInline):
     model = ZGWApiGroupConfig
@@ -67,8 +82,26 @@ class OpenZaakConfigAdmin(SingletonModelAdmin):
     )
 
 
+class ImportZGWDumpForm(Form):
+    MAX_UPLOAD_SIZE = 1024**2 * 25  # 25MB
+    zgw_export_file = LimitedUploadFileField(
+        label=_("ZGW export bestand"),
+        help_text=_(
+            "Upload a file generated by the export function on the 'Catalogus Config' administration page. The maximum size of this file is %(file_size)s"
+            % {"file_size": filesizeformat(MAX_UPLOAD_SIZE)}
+        ),
+        allow_empty_file=False,
+        required=True,
+        max_upload_size=MAX_UPLOAD_SIZE,
+        min_upload_size=1,
+        allowed_mime_types=["application/json", "text/plain"],
+    )
+
+
 @admin.register(CatalogusConfig)
 class CatalogusConfigAdmin(admin.ModelAdmin):
+    change_list_template = "admin/catalogusconfig_change_list.html"
+
     list_display = [
         "domein",
         "rsin",
@@ -90,6 +123,85 @@ class CatalogusConfigAdmin(admin.ModelAdmin):
     ordering = ("domein", "rsin")
     list_filter = ("service",)
 
+    actions = ["export_catalogus_configs"]
+
+    def get_urls(self):
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                "import-catalogus-dump/",
+                self.admin_site.admin_view(self.process_file_view),
+                name="upload_zgw_import_file",
+            ),
+        ]
+        return custom_urls + urls
+
+    @admin.action(description=_("Exporteer naar json export"))
+    def export_catalogus_configs(modeladmin, request, queryset):
+        export = CatalogusConfigExport.from_catalogus_configs(queryset)
+        response = StreamingHttpResponse(
+            export.as_jsonl_iter(),
+            content_type="application/json",
+        )
+        response[
+            "Content-Disposition"
+        ] = 'attachment; filename="zgw-catalogi-export.json"'
+        return response
+
+    def process_file_view(self, request):
+        form = None
+
+        match request.method:
+            case "POST":
+                form = ImportZGWDumpForm(request.POST, request.FILES)
+                if form.is_valid():
+                    storage = PrivateMediaFileSystemStorage()
+                    uploaded_file = request.FILES["zgw_export_file"]
+                    timestamp = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
+                    target_file_name = f"zgw_import_dump_{timestamp}.jsonl"
+                    storage.save(target_file_name, uploaded_file)
+
+                    try:
+                        import_result = CatalogusConfigImport.import_from_jsonl_file_in_django_storage(
+                            target_file_name,
+                            storage,
+                        )
+                        self.message_user(
+                            request,
+                            _(
+                                "Successfully processed %(num_rows)d items"
+                                % {"num_rows": import_result.total_rows_processed}
+                            ),
+                            messages.SUCCESS,
+                        )
+
+                        return HttpResponseRedirect(
+                            reverse(
+                                "admin:openzaak_catalogusconfig_changelist",
+                            )
+                        )
+                    except Exception as e:
+                        logger.exception("Unable to process ZGW import")
+                        self.message_user(
+                            request,
+                            _(
+                                "We were unable to process your upload. Please regenerate the file and try again."
+                            ),
+                            messages.ERROR,
+                        )
+                    finally:
+                        storage.delete(target_file_name)
+                else:
+                    self.message_user(
+                        request, _("No file was uploaded"), messages.WARNING
+                    )
+            case _:
+                form = ImportZGWDumpForm()
+
+        return TemplateResponse(
+            request, "admin/import_zgw_dump_form.html", {"form": form}
+        )
+
 
 class HasDocNotifyListFilter(admin.SimpleListFilter):
     title = _("notify document attachment")

src/open_inwoner/openzaak/tests/test_admin.py

@@ -1,12 +1,25 @@
+import json
+from unittest import mock
+
+from django.test import override_settings
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 
+import freezegun
 from django_webtest import WebTest
 from maykin_2fa.test import disable_admin_mfa
+from privates.storages import PrivateMediaFileSystemStorage
+from privates.test import temp_private_root
+from webtest import Upload
 
 from open_inwoner.accounts.tests.factories import UserFactory
 
-from .factories import ZaakTypeConfigFactory, ZaakTypeInformatieObjectTypeConfigFactory
+from .factories import (
+    CatalogusConfigFactory,
+    ServiceFactory,
+    ZaakTypeConfigFactory,
+    ZaakTypeInformatieObjectTypeConfigFactory,
+)
 
 
 @disable_admin_mfa()
@@ -109,3 +122,136 @@ def test_both_can_be_disabled(self):
         )
         self.assertFalse(self.ztc.document_upload_enabled)
         self.assertFalse(self.ztiotc.document_upload_enabled)
+
+
+@disable_admin_mfa()
+@temp_private_root()
+@freezegun.freeze_time("2024-08-14 17:50:01")
+@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
+class TestCatalogusConfigExportAdmin(WebTest):
+    csrf_checks = False
+
+    def setUp(self):
+        self.user = UserFactory(is_superuser=True, is_staff=True)
+        self.service = ServiceFactory(slug="service-1")
+        self.catalogus = CatalogusConfigFactory(
+            url="https://foo.maykinmedia.nl",
+            domein="DM",
+            rsin="123456789",
+            service=self.service,
+        )
+        self.mock_file = Upload("dump.json", b"foobar", "application/json")
+
+    def test_export_action_returns_correct_export(self):
+        response = self.app.post(
+            reverse(
+                "admin:openzaak_catalogusconfig_changelist",
+            ),
+            {
+                "action": "export_catalogus_configs",
+                "_selected_action": [self.catalogus.id],
+            },
+            user=self.user,
+        )
+
+        self.assertEqual(
+            response.content_disposition,
+            'attachment; filename="zgw-catalogi-export.json"',
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            [json.loads(row) for row in response.text.split("\n")[:-1]],
+            [
+                {
+                    "model": "openzaak.catalogusconfig",
+                    "fields": {
+                        "url": "https://foo.maykinmedia.nl",
+                        "domein": "DM",
+                        "rsin": "123456789",
+                        "service": ["service-1"],
+                    },
+                },
+            ],
+            msg="Response should be valid JSONL matching the input object",
+        )
+        self.assertFalse(
+            PrivateMediaFileSystemStorage().exists(
+                "zgw_import_dump_2024-08-14-17-50-01.jsonl"
+            ),
+            msg="File should always deleted regardless of success or failure",
+        )
+
+    @mock.patch(
+        "open_inwoner.openzaak.import_export.CatalogusConfigImport.import_from_jsonl_file_in_django_storage"
+    )
+    def test_import_flow_reports_success(self, m) -> None:
+        form = self.app.get(
+            reverse(
+                "admin:upload_zgw_import_file",
+            ),
+            user=self.user,
+        ).form
+        form["zgw_export_file"] = self.mock_file
+
+        response = form.submit().follow()
+
+        self.assertEqual(m.call_count, 1)
+        self.assertEqual(m.call_args[0][0], "zgw_import_dump_2024-08-14-17-50-01.jsonl")
+        self.assertTrue(isinstance(m.call_args[0][1], PrivateMediaFileSystemStorage))
+
+        messages = [str(msg) for msg in response.context["messages"]]
+        self.assertEqual(
+            messages,
+            ["Successfully processed 1 items"],
+        )
+        self.assertFalse(
+            PrivateMediaFileSystemStorage().exists(
+                "zgw_import_dump_2024-08-14-17-50-01.jsonl"
+            ),
+            msg="File should always deleted regardless of success or failure",
+        )
+        self.assertEqual(
+            response.request.path,
+            reverse(
+                "admin:openzaak_catalogusconfig_changelist",
+            ),
+        )
+
+    @mock.patch(
+        "open_inwoner.openzaak.import_export.CatalogusConfigImport.import_from_jsonl_file_in_django_storage"
+    )
+    def test_import_flow_errors_reports_failure_to_user(self, m) -> None:
+        m.side_effect = Exception("something went wrong")
+        form = self.app.get(
+            reverse(
+                "admin:upload_zgw_import_file",
+            ),
+            user=self.user,
+        ).form
+        form["zgw_export_file"] = self.mock_file
+
+        response = form.submit()
+
+        self.assertEqual(m.call_count, 1)
+        self.assertEqual(m.call_args[0][0], "zgw_import_dump_2024-08-14-17-50-01.jsonl")
+        self.assertTrue(isinstance(m.call_args[0][1], PrivateMediaFileSystemStorage))
+
+        messages = [str(msg) for msg in response.context["messages"]]
+        self.assertEqual(
+            messages,
+            [
+                "We were unable to process your upload. Please regenerate the file and try again."
+            ],
+        )
+        self.assertFalse(
+            PrivateMediaFileSystemStorage().exists(
+                "zgw_import_dump_2024-08-14-17-50-01.jsonl"
+            ),
+            msg="File should always deleted regardless of success or failure",
+        )
+        self.assertEqual(
+            response.request.path,
+            reverse(
+                "admin:upload_zgw_import_file",
+            ),
+        )

src/open_inwoner/templates/admin/catalogusconfig_change_list.html

@@ -0,0 +1,11 @@
+{% extends "admin/change_list.html" %}
+{% load i18n admin_urls %}
+
+{% block object-tools-items %}
+{{ block.super }}
+<li>
+    <a href="{% url 'admin:upload_zgw_import_file' %}" class="addlink">
+        {% trans "Create from json export" %}
+    </a>
+</li>
+{% endblock %}

src/open_inwoner/templates/admin/import_zgw_dump_form.html

@@ -0,0 +1,12 @@
+{% extends "admin/base_site.html" %}
+{% load i18n admin_urls %}
+
+{% block content %}
+<div id="main" class="main">
+    <form method="post" enctype="multipart/form-data">
+        {% csrf_token %}
+        {{ form.as_div }}
+        <input type="submit" value="Submit">
+    </form>
+</div>
+{% endblock %}