feat: extension model and form validation

geonode_mapstore_client/admin.py

@@ -1,7 +1,48 @@
+import os
+from django import forms
 from django.contrib import admin
-from geonode_mapstore_client.models import SearchService
+from geonode_mapstore_client.models import SearchService, Extension
 
 
 @admin.register(SearchService)
 class SearchServiceAdmin(admin.ModelAdmin):
     pass
+
+
+class ExtensionAdminForm(forms.ModelForm):
+    class Meta:
+        model = Extension
+        fields = '__all__'
+
+    def clean_uploaded_file(self):
+        """
+        It checks the uploaded file's name for uniqueness before the model is saved.
+        """
+        uploaded_file = self.cleaned_data.get('uploaded_file')
+
+        if uploaded_file:
+            extension_name = os.path.splitext(os.path.basename(uploaded_file.name))[0]
+
+            queryset = Extension.objects.filter(name=extension_name)
+
+            # If we are updating an existing instance, we can exclude it from the check
+            if self.instance.pk:
+                queryset = queryset.exclude(pk=self.instance.pk)
+
+            # If the queryset finds any conflicting extension, raise a validation error
+            if queryset.exists():
+                raise forms.ValidationError(
+                    f"An extension with the name '{extension_name}' already exists. Please upload a file with a different name."
+                )
+
+        return uploaded_file
+
+
+@admin.register(Extension)
+class ExtensionAdmin(admin.ModelAdmin):
+
+    form = ExtensionAdminForm
+    list_display = ('name', 'active', 'is_map_extension', 'updated_at')
+    list_filter = ('active', 'is_map_extension')
+    search_fields = ('name',)
+    readonly_fields = ('name', 'created_at', 'updated_at')

geonode_mapstore_client/apps.py

@@ -77,6 +77,9 @@ def run_setup_hooks(*args, **kwargs):
         pass
 
     urlpatterns += [
+        re_path("/client/extension", views.ExtensionsView.as_view(), name="mapstore-extension"),
+        re_path("/client/pluginsconfig", views.PluginsConfigView.as_view(), name="mapstore-pluginsconfig"),
+
         re_path(
             r"^catalogue/",
             TemplateView.as_view(

geonode_mapstore_client/migrations/0005_extension.py

@@ -0,0 +1,66 @@
+# Generated by Django 4.2.23 on 2025-10-03 12:14
+
+from django.db import migrations, models
+import geonode_mapstore_client.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("geonode_mapstore_client", "0004_auto_20231114_1705"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Extension",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "name",
+                    models.CharField(
+                        blank=True,
+                        help_text="Name of the extension, derived from the zip file name. Must be unique.",
+                        max_length=255,
+                        unique=True,
+                    ),
+                ),
+                (
+                    "uploaded_file",
+                    models.FileField(
+                        help_text="Upload the MapStore extension as a zip folder.",
+                        upload_to=geonode_mapstore_client.models.extension_upload_path,
+                        validators=[geonode_mapstore_client.models.validate_zip_file],
+                    ),
+                ),
+                (
+                    "active",
+                    models.BooleanField(
+                        default=True,
+                        help_text="Whether the extension is active and should be included in the index.",
+                    ),
+                ),
+                (
+                    "is_map_extension",
+                    models.BooleanField(
+                        default=False,
+                        help_text="Check if this extension is a map-specific plugin for Map Viewers.",
+                    ),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+            ],
+            options={
+                "verbose_name": "MapStore Extension",
+                "verbose_name_plural": "MapStore Extensions",
+                "ordering": ("name",),
+            },
+        ),
+    ]

geonode_mapstore_client/models.py

@@ -1,14 +1,21 @@
+import os
+import shutil
+import zipfile
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 from django.contrib.postgres.fields import ArrayField
 from django.dispatch import receiver
 from django.db.models import signals
 from django.core.cache import caches
-
+from django.db import models
+from geonode_mapstore_client.utils import validate_zip_file, clear_extension_caches
 from geonode_mapstore_client.templatetags.get_search_services import (
     populate_search_service_options,
 )
+from django.conf import settings
 
+# Define the target directory for static extensions
+EXTENSIONS_STATIC_DIR = os.path.join(settings.STATIC_ROOT, 'extensions')
 
 class SearchService(models.Model):
     class Meta:
@@ -73,3 +80,81 @@ def post_save_search_service(instance, sender, created, **kwargs):
         services_cache.delete("search_services")
 
     services_cache.set("search_services", populate_search_service_options(), 300)
+
+
+
+def extension_upload_path(instance, filename):
+    return f"mapstore_extensions/{filename}"
+
+
+class Extension(models.Model):
+    name = models.CharField(
+        max_length=255,
+        unique=True,
+        blank=True,  # Will be populated from the zip filename
+        help_text="Name of the extension, derived from the zip file name. Must be unique.",
+    )
+    uploaded_file = models.FileField(
+        upload_to=extension_upload_path,
+        validators=[validate_zip_file],
+        help_text="Upload the MapStore extension as a zip folder.",
+    )
+    active = models.BooleanField(
+        default=True,
+        help_text="Whether the extension is active and should be included in the index.",
+    )
+    is_map_extension = models.BooleanField(
+        default=False,
+        help_text="Check if this extension is a map-specific plugin for Map Viewers.",
+    )
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    def __str__(self):
+        return self.name
+
+    def save(self, *args, **kwargs):
+        if not self.name and self.uploaded_file:
+            self.name = os.path.splitext(os.path.basename(self.uploaded_file.name))[0]
+        super().save(*args, **kwargs)
+
+    class Meta:
+        ordering = ("name",)
+        verbose_name = "MapStore Extension"
+        verbose_name_plural = "MapStore Extensions"
+
+
+@receiver(signals.post_save, sender=Extension)
+def handle_extension_upload(sender, instance, **kwargs):
+    """
+    Unzips the extension file and clears the API cache after saving.
+    """
+    os.makedirs(EXTENSIONS_STATIC_DIR, exist_ok=True)
+    target_path = os.path.join(EXTENSIONS_STATIC_DIR, instance.name)
+
+    if os.path.exists(target_path):
+        shutil.rmtree(target_path)
+
+    try:
+        with zipfile.ZipFile(instance.uploaded_file.path, "r") as zip_ref:
+            zip_ref.extractall(target_path)
+    except FileNotFoundError:
+        pass
+
+    clear_extension_caches()
+
+
+@receiver(signals.post_delete, sender=Extension)
+def handle_extension_delete(sender, instance, **kwargs):
+    """
+    Removes the extension's files and clears the API cache on deletion.
+    """
+    if instance.name:
+        extension_path = os.path.join(EXTENSIONS_STATIC_DIR, instance.name)
+        if os.path.exists(extension_path):
+            shutil.rmtree(extension_path)
+
+    if instance.uploaded_file and os.path.exists(instance.uploaded_file.path):
+        os.remove(instance.uploaded_file.path)
+
+    clear_extension_caches()

geonode_mapstore_client/tests.py

@@ -0,0 +1,156 @@
+import os
+import shutil
+import zipfile
+from io import BytesIO
+
+from django.conf import settings
+from django.core.cache import cache
+from django.core.exceptions import ValidationError
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase, override_settings
+from django.urls import reverse
+from rest_framework.test import APIClient
+
+from .utils import validate_zip_file
+from .admin import ExtensionAdminForm
+from .models import Extension
+
+# Define temporary directories for testing to avoid affecting the real media/static roots
+TEST_MEDIA_ROOT = os.path.join(settings.PROJECT_ROOT, 'test_media')
+TEST_STATIC_ROOT = os.path.join(settings.PROJECT_ROOT, 'test_static')
+
+
+@override_settings(MEDIA_ROOT=TEST_MEDIA_ROOT, STATIC_ROOT=TEST_STATIC_ROOT)
+class ExtensionFeatureTestCase(TestCase):
+    """
+    A comprehensive test case for the MapStore Extension feature, updated to match
+    the latest code with constants and new API response structures.
+    """
+
+    def setUp(self):
+        """Set up the test environment."""
+        self.tearDown()
+        os.makedirs(TEST_MEDIA_ROOT, exist_ok=True)
+        os.makedirs(TEST_STATIC_ROOT, exist_ok=True)
+        self.client = APIClient()
+        cache.clear()
+
+    def tearDown(self):
+        """Clean up the test directories after each test."""
+        if os.path.exists(TEST_MEDIA_ROOT):
+            shutil.rmtree(TEST_MEDIA_ROOT)
+        if os.path.exists(TEST_STATIC_ROOT):
+            shutil.rmtree(TEST_STATIC_ROOT)
+
+    def _create_mock_zip_file(self, filename="SampleExtension.zip", add_index_js=True, add_index_json=True):
+        """Creates an in-memory zip file for testing uploads."""
+        zip_buffer = BytesIO()
+        with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
+            if add_index_js:
+                zf.writestr('index.js', 'console.log("hello");')
+            if add_index_json:
+                zf.writestr('index.json', '{"name": "test"}')
+        zip_buffer.seek(0)
+        return SimpleUploadedFile(filename, zip_buffer.read(), content_type='application/zip')
+
+    def test_model_save_derives_name_from_file(self):
+        """Test that the Extension.save() method correctly sets the name."""
+        mock_zip = self._create_mock_zip_file()
+        ext = Extension.objects.create(uploaded_file=mock_zip)
+        self.assertEqual(ext.name, "SampleExtension")
+
+    def test_form_prevents_duplicate_names(self):
+        """Test that ExtensionAdminForm validation fails for a duplicate name."""
+        Extension.objects.create(uploaded_file=self._create_mock_zip_file())
+        form_data = {}
+        file_data = {'uploaded_file': self._create_mock_zip_file()}
+        form = ExtensionAdminForm(data=form_data, files=file_data)
+        self.assertFalse(form.is_valid())
+        self.assertIn('uploaded_file', form.errors)
+        self.assertIn("already exists", form.errors['uploaded_file'][0])
+
+    def test_zip_validator_raises_error_for_invalid_file(self):
+        """Test that validate_zip_file raises an error for non-zip files."""
+        invalid_file = SimpleUploadedFile("test.txt", b"not a zip file")
+        with self.assertRaises(ValidationError) as context:
+            validate_zip_file(invalid_file)
+        self.assertIn("not a valid zip archive", str(context.exception))
+
+    def test_zip_validator_raises_error_for_missing_required_files(self):
+        """Test that validate_zip_file fails if index.js or index.json is missing."""
+        missing_js_zip = self._create_mock_zip_file(add_index_js=False)
+        with self.assertRaises(ValidationError) as context:
+            validate_zip_file(missing_js_zip)
+        self.assertIn("must contain index.js and index.json", str(context.exception))
+
+
+    def test_post_save_signal_unzips_file_and_clears_cache(self):
+        """Test that the post_save signal unzips the file and clears the cache."""
+        ext = Extension.objects.create(uploaded_file=self._create_mock_zip_file())
+        self.assertEqual(ext.name, "SampleExtension")
+
+        expected_dir = os.path.join(TEST_STATIC_ROOT, 'extensions', ext.name)
+
+        self.assertTrue(os.path.isdir(expected_dir), f"Directory {expected_dir} was not created.")
+        self.assertTrue(os.path.exists(os.path.join(expected_dir, 'index.js')))
+
+
+    def test_post_delete_signal_removes_files_and_clears_cache(self):
+        """Test that the post_delete signal removes files and clears the cache."""
+        ext = Extension.objects.create(uploaded_file=self._create_mock_zip_file())
+        zip_path = ext.uploaded_file.path
+        unzipped_dir = os.path.join(TEST_STATIC_ROOT, 'extensions', ext.name)
+        self.assertTrue(os.path.exists(zip_path))
+        self.assertTrue(os.path.isdir(unzipped_dir))
+        ext.delete()
+        self.assertFalse(os.path.exists(zip_path))
+        self.assertFalse(os.path.isdir(unzipped_dir))
+
+    def test_extensions_view(self):
+        """Test the extensions index API endpoint."""
+        Extension.objects.create(name="ActiveExt", active=True)
+        Extension.objects.create(name="InactiveExt", active=False)
+
+        mock_legacy_dir = os.path.join(TEST_STATIC_ROOT, 'mapstore', 'extensions')
+        os.makedirs(mock_legacy_dir, exist_ok=True)
+        with open(os.path.join(mock_legacy_dir, 'index.json'), 'w') as f:
+            f.write('{"LegacyExt": {"bundle": "/static/legacy/bundle.js"}}')
+
+        url = reverse('mapstore-extension')
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+
+        self.assertIn("ActiveExt", data)
+        self.assertIn("LegacyExt", data)
+        self.assertNotIn("InactiveExt", data)
+
+    def test_plugins_config_view_structure(self):
+        """Test the plugins config API endpoint and its new response structure."""
+        mock_file = self._create_mock_zip_file()
+        Extension.objects.create(name="MapPlugin", active=True, is_map_extension=True, uploaded_file=mock_file)
+        Extension.objects.create(name="NotAMapPlugin", active=True, is_map_extension=False, uploaded_file=mock_file)
+
+        url = reverse('mapstore-pluginsconfig')
+
+        mock_config_dir = os.path.join(settings.PROJECT_ROOT, 'static', 'mapstore', 'configs')
+        os.makedirs(mock_config_dir, exist_ok=True)
+        with open(os.path.join(mock_config_dir, 'pluginsConfig.json'), 'w') as f:
+            f.write('{"plugins": [{"name": "BasePlugin"}]}')
+
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertIn("plugins", data)
+
+        plugin_list = data['plugins']
+        plugin_names = {p.get('name') for p in plugin_list}
+
+        self.assertIn("MapPlugin", plugin_names)
+        self.assertIn("BasePlugin", plugin_names)
+        self.assertNotIn("NotAMapPlugin", plugin_names)
+
+        map_plugin_data = next((p for p in plugin_list if p.get('name') == "MapPlugin"), None)
+        self.assertIsNotNone(map_plugin_data)
+        self.assertIn("bundle", map_plugin_data)
+        self.assertTrue(map_plugin_data['bundle'].endswith('/extensions/MapPlugin/index.js'))