Add button to clear cache in the admin page (#4390)

bims/api_urls.py

@@ -1,6 +1,7 @@
 from django.urls import re_path, path
 
 from bims.api_views.checklist import DownloadChecklistAPIView
+from bims.api_views.clear_cache import ClearCacheView
 from bims.api_views.geocontext import (
     IsHarvestingGeocontext, HarvestGeocontextView, ClearHarvestingGeocontextCache,
     GetGeocontextLogLinesView
@@ -408,4 +409,7 @@
     path('cloud-native-layers/',
          CloudNativeLayerList.as_view(),
          name='cloud-native-layers'),
+    path('clear-cache/',
+         ClearCacheView.as_view(),
+         name='clear_cache_view'),
 ]

bims/api_views/clear_cache.py

@@ -0,0 +1,25 @@
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from django.core.cache import cache
+from rest_framework.status import HTTP_200_OK, HTTP_403_FORBIDDEN
+
+from bims.api_views.merge_sites import IsSuperUser
+from bims.models.search_process import SearchProcess
+
+
+class ClearCacheView(APIView):
+    permission_classes = (IsSuperUser,)
+
+    def post(self, request, *args, **kwargs):
+        if request.user.is_superuser:
+            cache.clear()
+            SearchProcess.objects.filter(
+                finished=True,
+                locked=False
+            ).delete()
+            return Response(
+                {"message": "Cache cleared successfully."},
+                status=HTTP_200_OK)
+        return Response(
+            {"error": "Permission denied."},
+            status=HTTP_403_FORBIDDEN)

bims/templates/admin/index.html

@@ -0,0 +1,58 @@
+{% extends "admin/index.html" %}
+
+{% block content %}
+{{ block.super }}
+
+<script>
+    document.addEventListener("DOMContentLoaded", function () {
+        const targetElement = document.querySelector('.g-d-c');
+        if (targetElement) {
+            // Create the button
+            const customButton = document.createElement('a');
+            customButton.href = "#";
+            customButton.className = "grp-button grp-primary";
+            customButton.textContent = "Clear Cache";
+            customButton.style.marginLeft = "10px";
+
+            // Add click event with confirmation
+            customButton.addEventListener("click", function (event) {
+                event.preventDefault();
+
+                // Confirmation dialog
+                if (confirm("Are you sure you want to clear the cache?")) {
+                    // Make an AJAX request
+                    fetch("{% url 'clear_cache_view' %}", {
+                        method: "POST",
+                        headers: {
+                            "X-CSRFToken": "{{ csrf_token }}",
+                            "Content-Type": "application/json"
+                        },
+                        credentials: 'same-origin'
+                    })
+                    .then(response => {
+                        if (response.ok) {
+                            return response.json();
+                        } else {
+                            throw new Error("Failed to clear cache");
+                        }
+                    })
+                    .then(data => {
+                        alert(data.message);
+                        window.location.href = "/admin/";
+                    })
+                    .catch(error => {
+                        console.error("Error clearing cache:", error);
+                        alert("An unexpected error occurred. Please try again.");
+                    });
+                }
+            });
+
+            // Append the button to the target element
+            targetElement.appendChild(customButton);
+        } else {
+            console.warn("Element with class 'g-d-c' not found.");
+        }
+    });
+</script>
+
+{% endblock %}

bims/tests/test_clear_cache.py

@@ -0,0 +1,89 @@
+
+from django.core.cache import cache
+from django_tenants.test.cases import FastTenantTestCase
+from django_tenants.test.client import TenantClient
+
+from bims.models.search_process import SearchProcess
+from rest_framework.status import HTTP_200_OK, HTTP_403_FORBIDDEN, HTTP_401_UNAUTHORIZED
+
+from bims.tests.model_factories import UserF
+
+
+class ClearCacheViewTest(FastTenantTestCase):
+
+    def setUp(self):
+        # Call super to set up the tenant
+        super().setUp()
+
+        # Create a superuser
+        self.superuser = UserF.create(
+            is_superuser=True
+        )
+
+        # Create a regular user
+        self.user = UserF.create(
+            username="user",
+            password="password",
+        )
+
+        # Create the tenant-specific client
+        self.client = TenantClient(self.tenant)
+
+        # Create sample SearchProcess objects
+        SearchProcess.objects.create(finished=True, locked=False)
+        SearchProcess.objects.create(finished=True, locked=True)
+
+        # Populate the cache
+        cache.set("test_key", "test_value")
+
+    def test_clear_cache_superuser(self):
+        # Authenticate as the superuser
+        self.client.login(username=self.superuser.username, password="password")
+
+        # Make the POST request
+        response = self.client.post("/api/clear-cache/")
+
+        # Assertions
+        self.assertEqual(response.status_code, HTTP_200_OK)
+        self.assertEqual(response.json(), {"message": "Cache cleared successfully."})
+
+        # Verify cache is cleared
+        self.assertIsNone(cache.get("test_key"))
+
+        # Verify finished and unlocked SearchProcess objects are deleted
+        self.assertEqual(SearchProcess.objects.filter(finished=True, locked=False).count(), 0)
+
+        # Verify finished and locked SearchProcess objects remain
+        self.assertEqual(SearchProcess.objects.filter(finished=True, locked=True).count(), 1)
+
+    def test_clear_cache_non_superuser(self):
+        # Authenticate as a regular user
+        self.client.login(username="user", password="password")
+
+        # Make the POST request
+        response = self.client.post("/api/clear-cache/")
+
+        # Assertions
+        self.assertEqual(response.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(response.json(), {"detail": "You do not have permission to perform this action."})
+
+        # Verify cache is not cleared
+        self.assertEqual(cache.get("test_key"), "test_value")
+
+    def test_clear_cache_unauthenticated(self):
+        # Make the POST request without authentication
+        response = self.client.post("/api/clear-cache/")
+
+        # Assertions
+        self.assertEqual(response.status_code, HTTP_401_UNAUTHORIZED)
+        self.assertEqual(response.json(), {"detail": "Authentication credentials were not provided."})
+
+    def test_clear_cache_invalid_method(self):
+        # Authenticate as the superuser
+        self.client.login(username="admin", password="password")
+
+        # Make a GET request
+        response = self.client.get("/api/clear-cache/")
+
+        # Assertions
+        self.assertEqual(response.status_code, HTTP_401_UNAUTHORIZED)