diff --git a/program/models.py b/program/models.py
index 9898b108a14f9688adfdd33e4fc5cdfbadfd586d..87dfa525287451cb0739578f09656ee466c0ea5c 100644
--- a/program/models.py
+++ b/program/models.py
@@ -17,8 +17,10 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
+import collections
import dataclasses
import datetime
+import typing
import jsonschema
from rest_framework.exceptions import ValidationError
@@ -705,3 +707,71 @@ class ProgramEntry:
return self.timeslot.playlist_id
else:
return self.show.default_playlist_id
+
+
+class ApplicationStateManager:
+ categorized_models: typing.Mapping[str, list[typing.Type[models.Model]]] = {
+ "classifications": [
+ Type,
+ Category,
+ Topic,
+ MusicFocus,
+ FundingCategory,
+ Language,
+ License,
+ LinkType,
+ RRule,
+ ],
+ "settings": [RadioSettings],
+ "auth": [CBA, User],
+ "media": [Image],
+ "program": [Note, TimeSlot, Schedule, Show],
+ }
+
+ @property
+ def _models(self):
+ result = []
+ for _models in self.categorized_models.values():
+ result.extend(_models)
+ return result
+
+ @property
+ def _model_map(self):
+ return {model._meta.label: model for model in self._models}
+
+ @property
+ def model_category_choices(self):
+ return sorted(self.categorized_models.keys())
+
+ @property
+ def model_choices(self):
+ return sorted(self._model_map.keys())
+
+ def purge(
+ self,
+ model_category_names: typing.Iterable[str] | None = None,
+ model_names: typing.Iterable[str] | None = None,
+ invert_selection: bool = False,
+ ):
+ model_map = self._model_map
+ model_category_names = set(model_category_names or [])
+ model_names = set(model_names or [])
+
+ selected_models: set[typing.Type[models.Model]] = set()
+ for category_name in model_category_names:
+ selected_models.update(self.categorized_models[category_name])
+ for model_name in model_names:
+ selected_models.add(model_map[model_name])
+ if invert_selection:
+ selected_models = set(self._models).difference(selected_models)
+
+ # Some models may have dependent state and therefore need to be deleted in order.
+ ordered_selected_models = [model for model in self._models if model in selected_models]
+ deleted_model_count_map = collections.defaultdict(int)
+ for model in ordered_selected_models:
+ for model_path, count in model.objects.all().delete()[1].items():
+ deleted_model_count_map[model_path] += count
+ return deleted_model_count_map
+
+
+application_state_manager = ApplicationStateManager()
diff --git a/program/serializers.py b/program/serializers.py
index e34aa9b23645c20943c021a2a37940f70136abb6..bbae2a63899d1d5f737b7ea5781927bc993baf73 100644
--- a/program/serializers.py
+++ b/program/serializers.py
@@ -52,6 +52,7 @@ from program.models import (
TimeSlot,
Topic,
Type,
+ application_state_manager,
)
from program.typing import (
Logo,
@@ -1399,3 +1400,34 @@ class CalendarSchemaSerializer(serializers.Serializer):
episodes = CalendarEpisodeSerializer(many=True)
licenses = LicenseSerializer(many=True)
link_types = LinkTypeSerializer(many=True)
+
+
+class ApplicationStatePurgeSerializer(serializers.Serializer):
+ @staticmethod
+ def _render_model_category_definitions():
+ yield "<dl>"
+ for category_name, models in application_state_manager.categorized_models.items():
+ model_names = ", ".join(sorted(model._meta.label for model in models))
+ yield f"<dt>{category_name}</dt>"
+ yield f"<dd>{model_names}</dd>"
+ yield "</dl>"
+
+ models = serializers.MultipleChoiceField(
+ choices=application_state_manager.model_choices, default=set()
+ )
+ model_categories = serializers.MultipleChoiceField(
+ choices=application_state_manager.model_category_choices,
+ default=set(),
+ help_text=(
+ "Selects multiple models by their categorization. "
+ "Models included in the categories are: "
+ f"{''.join(_render_model_category_definitions())}"
+ ),
+ )
+ invert_selection = serializers.BooleanField(
+ default=False,
+ help_text=(
+ "Inverts the model selection that is selected through other filters. "
+ "Selects all models if set to true and no other filters have been set."
+ ),
+ )
diff --git a/program/views.py b/program/views.py
index c970cbc4c11be93d0aef298f6d1c6bd3371c79b1..a03556f645a8bf84dc04fd5682b2550211b4a526 100644
--- a/program/views.py
+++ b/program/views.py
@@ -33,8 +33,9 @@ from drf_spectacular.utils import (
)
from rest_framework import decorators
from rest_framework import filters as drf_filters
-from rest_framework import mixins, permissions, status, viewsets
+from rest_framework import mixins, permissions, status, views, viewsets
from rest_framework.pagination import LimitOffsetPagination
+from rest_framework.request import Request
from rest_framework.response import Response
from django.conf import settings
@@ -62,8 +63,10 @@ from program.models import (
TimeSlot,
Topic,
Type,
+ application_state_manager,
)
from program.serializers import (
+ ApplicationStatePurgeSerializer,
BasicProgramEntrySerializer,
CalendarSchemaSerializer,
CategorySerializer,
@@ -1281,3 +1284,46 @@ class APILicenseViewSet(viewsets.ModelViewSet):
class APIRadioSettingsViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
queryset = RadioSettings.objects.all()
serializer_class = RadioSettingsSerializer
+
+
+class TestOperationPermission(permissions.BasePermission):
+ def has_permission(self, request, view):
+ return settings.OPERATION_MODE == "tests"
+
+ def has_object_permission(self, request, view, obj):
+ return self.has_permission(request, view)
+
+
+class TestOperationViewMixin:
+ authentication_classes = []
+ permission_classes = [TestOperationPermission]
+
+
+class APIApplicationStateView(TestOperationViewMixin, views.APIView):
+ @extend_schema(
+ description=(
+ "Purges any data of the selected models and "
+ "returns the deleted count for each of them."
+ ),
+ parameters=[ApplicationStatePurgeSerializer],
+ responses={200: dict[str, int]},
+ examples=[
+ OpenApiExample(
+ "Example entry",
+ value={
+ "program.Show": 6,
+ "program.TimeSlot": 38,
+ },
+ ),
+ ],
+ tags=["debug"],
+ )
+ def delete(self, request: Request, *args, **kwargs):
+ params = ApplicationStatePurgeSerializer(data=request.query_params)
+ params.is_valid(raise_exception=True)
+ deleted = application_state_manager.purge(
+ model_category_names=set(params.validated_data["model_categories"]),
+ model_names=set(params.validated_data["models"]),
+ invert_selection=params.validated_data["invert_selection"],
+ )
+ return Response(status=status.HTTP_200_OK, data=deleted)
diff --git a/steering/settings.py b/steering/settings.py
index 773bf3ca2e0bd45c053572853bc61075b0b6ccd1..c77936fb9df2fa6aa094f962effc6e442b6553e0 100644
--- a/steering/settings.py
+++ b/steering/settings.py
@@ -1,6 +1,7 @@
import os
import sys
from pathlib import Path
+from typing import Literal, cast
import ldap
from corsheaders.defaults import default_headers
@@ -218,6 +219,10 @@ SITE_URL = f"{AURA_PROTO}://{AURA_HOST}:{PORT}" if PORT else f"{AURA_PROTO}://{A
if AURA_PROTO == "https":
CSRF_TRUSTED_ORIGINS = [f"{AURA_PROTO}://{AURA_HOST}"]
+OPERATION_MODE_TYPE = Literal["default", "tests"]
+OPERATION_MODE: OPERATION_MODE_TYPE
+OPERATION_MODE = cast(OPERATION_MODE_TYPE, os.getenv("OPERATION_MODE", default="default"))
+
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
diff --git a/steering/urls.py b/steering/urls.py
index ccf2a0d0f7a65d60e2504ab54b8a5235d3ea6f1c..d863f58424d8ae9f88f2326d531e5b461d8b9d89 100644
--- a/steering/urls.py
+++ b/steering/urls.py
@@ -26,6 +26,7 @@ from django.conf.urls.static import static
from django.contrib import admin
from django.urls import include, path
from program.views import (
+ APIApplicationStateView,
APICategoryViewSet,
APIFundingCategoryViewSet,
APIImageViewSet,
@@ -83,3 +84,10 @@ urlpatterns = [
),
path("admin/", admin.site.urls),
] + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
+
+if settings.DEBUG or settings.OPERATION_MODE == "tests":
+ # This endpoint is only usable with OPERATION_MODE == "tests"
+ # but we include it in DEBUG mode so that developers can inspect the schema.
+ urlpatterns.append(
+ path("api/v1/debug/application-state/", APIApplicationStateView.as_view()),
+ )