[Fixes #8784] Implement recalculation of Geoserver featuretype and coverage bbox and update of dataset bbox (#13768)

* adding an endpoint and a helper for the bbox recalculation of GeoServer and GeoNode

* move the part of REST in the geoserver-restconfig library

* using  _to_compact_perms to check user perms

Author: Gpetrak

geonode/layers/api/views.py

@@ -36,7 +36,10 @@
 from geonode.metadata.multilang.views import MultiLangViewMixin
 from geonode.resource.utils import update_resource
 from geonode.resource.manager import resource_manager
-from rest_framework.exceptions import NotFound
+from geonode.security.registry import permissions_registry
+from geonode.security.permissions import _to_compact_perms
+
+from rest_framework.exceptions import NotFound, PermissionDenied
 from django.shortcuts import get_object_or_404
 from django.http import JsonResponse
 
@@ -292,3 +295,28 @@ def timeseries_info(self, request, pk, *args, **kwards):
                 layer.save()
 
                 return JsonResponse({"message": "The time dimension information for this layer was disabled"})
+
+    @action(
+        detail=True,
+        methods=["put"],
+        url_path="recalc-bbox",
+        url_name="recalc_bbox",
+        permission_classes=[IsAuthenticated],
+    )
+    def recalc_bbox(self, request, pk=None, *args, **kwargs):
+        dataset = self.get_object()
+
+        # Permissions check
+        access = _to_compact_perms(permissions_registry.get_perms(instance=dataset, user=request.user))
+        if access not in ("owner", "edit", "manage"):
+            raise PermissionDenied("You do not have permission to edit this dataset")
+
+        # Safely get bbox from request
+        force_bbox = getattr(request, "data", {}) or {}
+        force_bbox = force_bbox.get("bbox", None)
+
+        success = dataset.recalc_bbox_on_geoserver(force_bbox=force_bbox)
+
+        if success:
+            return Response({"success": True})
+        return Response({"success": False}, status=500)

geonode/layers/models.py

@@ -21,7 +21,7 @@
 import logging
 
 from django.conf import settings
-from django.db import models
+from django.db import models, transaction
 from django.urls import reverse
 from django.utils.timezone import now
 from django.utils.functional import classproperty
@@ -306,6 +306,57 @@ def attribute_config(self):
 
         return cfg
 
+    def recalc_bbox_on_geoserver(self, force_bbox=None):
+        """
+        Delegate BBOX recalculation/update to the GeoServer layer object,
+        then refresh the Dataset's bbox fields from the updated resource.
+
+        This wraps the Layer.recalc_bbox() method (works for both raster + vector).
+        """
+
+        from geonode.geoserver.helpers import gs_catalog
+
+        # GeoServer layer object (has .resource)
+        gs_layer = gs_catalog.get_layer(self.name)
+        if gs_layer is None:
+            logger.error(f"GeoServer layer not found: {self.name}")
+            return False
+
+        # Call recalc_bbox method from the geoserver-restconfig library
+        ok = gs_layer.recalc_bbox(force_bbox=force_bbox)
+        if not ok:
+            logger.error(f"GeoServer refused bbox update for layer {self.name}")
+            return False
+
+        # Let's reset the connections first
+        gs_catalog._cache.clear()
+        gs_catalog.reset()
+        # Fetch the updated resource again from GeoServer
+        resource = gs_catalog.get_resource(
+            name=self.name, store=gs_layer.resource.store, workspace=gs_layer.resource.workspace
+        )
+
+        if not resource:
+            logger.error("No resource returned from GeoServer after bbox update")
+            return False
+
+        bbox = resource.native_bbox
+        ll = resource.latlon_bbox
+        srid = resource.projection
+
+        if not bbox or not ll:
+            logger.error("GeoServer did not return updated bbox values")
+            return False
+
+        # bbox order from GeoServer: [minx, maxx, miny, maxy]
+        with transaction.atomic():
+            self.set_bbox_polygon([bbox[0], bbox[2], bbox[1], bbox[3]], srid)
+            self.set_ll_bbox_polygon([ll[0], ll[2], ll[1], ll[3]])
+            self.srid = srid or self.srid
+            self.save(update_fields=["srid"])
+
+        return True
+
     def __str__(self):
         return str(self.alternate)
 

geonode/layers/tests.py

@@ -26,6 +26,9 @@
 from unittest.mock import MagicMock, patch, PropertyMock
 from collections import namedtuple
 
+from rest_framework.test import force_authenticate
+from rest_framework import status
+
 from django.urls import reverse
 from django.test import TestCase
 from django.test.client import RequestFactory
@@ -799,6 +802,109 @@ def test_supports_time_with_raster_subtype_and_invalid_attributes(self, mock_get
         mock_dataset = Dataset(subtype="raster")
         self.assertFalse(mock_dataset.supports_time)
 
+    @patch("geonode.geoserver.helpers.gs_catalog.get_resource")
+    @patch("geonode.geoserver.helpers.gs_catalog.get_layer")
+    def test_dataset_recalc_bbox_on_geoserver_success(self, mock_get_layer, mock_get_resource):
+        """
+        Test recalc_bbox_on_geoserver() without touching real GeoServer.
+        """
+
+        # Mock the GeoServer Layer object
+        mock_layer = MagicMock()
+        mock_layer.resource.store = MagicMock(name="test_store", workspace=MagicMock(name="test_workspace"))
+        # Mock recalc_bbox() returning True
+        mock_layer.recalc_bbox.return_value = True
+        mock_get_layer.return_value = mock_layer
+
+        # Mock the resource returned by gs_catalog.get_resource
+        mock_resource = MagicMock()
+        mock_resource.native_bbox = [0, 10, 0, 10]
+        mock_resource.latlon_bbox = [0, 10, 0, 10]
+        mock_resource.projection = "EPSG:4326"
+        mock_resource.store = mock_layer.resource.store
+        mock_get_resource.return_value = mock_resource
+
+        # Call the method
+        result = self.dataset.recalc_bbox_on_geoserver(force_bbox=[0, 0, 10, 10])
+
+        # Assertions
+        self.assertTrue(result)
+        mock_layer.recalc_bbox.assert_called_once_with(force_bbox=[0, 0, 10, 10])
+        mock_get_resource.assert_called_once_with(
+            name=self.dataset.name, store=mock_layer.resource.store, workspace=mock_layer.resource.workspace
+        )
+
+        # Check if bbox and srid are updated correctly
+        self.assertEqual(self.dataset.srid, "EPSG:4326")
+        # Optionally, check polygons
+        self.assertIsNotNone(self.dataset.bbox_polygon)
+        self.assertIsNotNone(self.dataset.ll_bbox_polygon)
+
+    @patch("geonode.layers.models.Dataset.recalc_bbox_on_geoserver")
+    def test_recalc_bbox_view_success(self, mock_recalc_bbox):
+        """Test the recalc_bbox view returns 200 when successful."""
+        factory = RequestFactory()
+        django_request = factory.put(
+            f"/api/v2/datasets/{self.dataset.id}/recalc-bbox/",
+            data='{"bbox": [0, 0, 10, 10]}',
+            content_type="application/json",
+        )
+        force_authenticate(django_request, user=self.admin_user)  # authenticate the HttpRequest
+
+        mock_recalc_bbox.return_value = True
+
+        from geonode.layers.api.views import DatasetViewSet
+
+        view = DatasetViewSet.as_view({"put": "recalc_bbox"})
+        response = view(django_request, pk=self.dataset.id)  # pass HttpRequest directly
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data == {"success": True}
+        mock_recalc_bbox.assert_called_once_with(force_bbox=[0, 0, 10, 10])
+
+    @patch("geonode.layers.models.Dataset.recalc_bbox_on_geoserver")
+    def test_recalc_bbox_view_failure(self, mock_recalc_bbox):
+        """Test the recalc_bbox view returns 500 when GeoServer update fails."""
+        factory = RequestFactory()
+        django_request = factory.put(
+            f"/api/v2/datasets/{self.dataset.id}/recalc-bbox/",
+            data='{"bbox": [0, 0, 10, 10]}',
+            content_type="application/json",
+        )
+        force_authenticate(django_request, user=self.admin_user)
+
+        mock_recalc_bbox.return_value = False
+
+        from geonode.layers.api.views import DatasetViewSet
+
+        view = DatasetViewSet.as_view({"put": "recalc_bbox"})
+        response = view(django_request, pk=self.dataset.id)
+
+        assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+        assert response.data == {"success": False}
+        mock_recalc_bbox.assert_called_once_with(force_bbox=[0, 0, 10, 10])
+
+    @patch("geonode.layers.models.Dataset.recalc_bbox_on_geoserver")
+    def test_recalc_bbox_view_permission_denied(self, mock_recalc_bbox):
+        """Test the recalc_bbox view denies access if user lacks edit perms."""
+        factory = RequestFactory()
+        django_request = factory.put(
+            f"/api/v2/datasets/{self.dataset.id}/recalc-bbox/",
+            data='{"bbox": [0, 0, 10, 10]}',
+            content_type="application/json",
+        )
+        force_authenticate(django_request, user=self.anonymous_user)  # no edit perms
+
+        from geonode.layers.api.views import DatasetViewSet
+
+        view = DatasetViewSet.as_view({"put": "recalc_bbox"})
+
+        response = view(django_request, pk=self.dataset.id)
+
+        # DRF converts PermissionDenied to HTTP 403 response
+        assert response.status_code == 403
+        mock_recalc_bbox.assert_not_called()
+
 
 class TestLayerDetailMapViewRights(GeoNodeBaseTestSupport):
     fixtures = ["initial_data.json", "group_test_data.json", "default_oauth_apps.json"]