add "clean up events" button to EventLog admin (#84)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: joshuadavidthomas

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project attempts to adhere to [Semantic Versioning](https://semver.org/
 ### Added
 
 - Added `GITHUB_APP["LOG_ALL_EVENTS"]` setting to control webhook event logging. When `False`, only events with registered handlers are stored in the database.
+- Added admin action to bulk delete EventLog entries older than a specified number of days.
 
 ## [0.6.1]
 

src/django_github_app/admin.py

@@ -1,17 +1,108 @@
 from __future__ import annotations
 
+import datetime
+from collections.abc import Sequence
+
+from django import forms
 from django.contrib import admin
+from django.contrib import messages
+from django.core.exceptions import ValidationError
+from django.http import HttpRequest
+from django.http import HttpResponse
+from django.http import HttpResponseRedirect
+from django.shortcuts import render
+from django.urls import URLPattern
+from django.urls import URLResolver
+from django.urls import path
+from django.urls import reverse
+from django.utils import timezone
 
+from ._typing import override
+from .conf import app_settings
 from .models import EventLog
 from .models import Installation
 from .models import Repository
 
 
+class EventLogCleanupForm(forms.Form):
+    days_to_keep = forms.IntegerField(
+        label="Days to keep",
+        min_value=0,
+        initial=app_settings.DAYS_TO_KEEP_EVENTS,
+        help_text="Event logs older than this number of days will be deleted.",
+    )
+
+    def save(self) -> int:
+        """Delete the events and return the count."""
+        days_to_keep = self.cleaned_data["days_to_keep"]
+        deleted_count, _ = EventLog.objects.cleanup_events(days_to_keep)
+        return deleted_count
+
+    @property
+    def to_delete_count(self) -> int:
+        if not hasattr(self, "cleaned_data"):  # pragma: no cover
+            raise ValidationError(
+                "Form must be validated before accessing to_delete_count"
+            )
+        return EventLog.objects.filter(received_at__lte=self.cutoff_date).count()
+
+    @property
+    def cutoff_date(self) -> datetime.datetime:
+        if not hasattr(self, "cleaned_data"):  # pragma: no cover
+            raise ValidationError("Form must be validated before accessing cutoff_date")
+        days_to_keep = self.cleaned_data["days_to_keep"]
+        return timezone.now() - datetime.timedelta(days=days_to_keep)
+
+
 @admin.register(EventLog)
 class EventLogModelAdmin(admin.ModelAdmin):
     list_display = ["id", "event", "action", "received_at"]
     readonly_fields = ["event", "payload", "received_at"]
 
+    @override
+    def get_urls(self) -> Sequence[URLResolver | URLPattern]:  # type: ignore[override]
+        urls = super().get_urls()
+        custom_urls = [
+            path(
+                "cleanup/",
+                self.admin_site.admin_view(self.cleanup_view),
+                name="django_github_app_eventlog_cleanup",
+            ),
+        ]
+        return custom_urls + urls
+
+    def cleanup_view(self, request: HttpRequest) -> HttpResponse:
+        form = EventLogCleanupForm(request.POST or None)
+
+        # handle confirmation
+        if request.POST.get("post") == "yes" and form.is_valid():
+            deleted_count = form.save()
+            days_to_keep = form.cleaned_data["days_to_keep"]
+            event_text = "event" if deleted_count == 1 else "events"
+            day_text = "day" if days_to_keep == 1 else "days"
+            messages.success(
+                request,
+                f"Successfully deleted {deleted_count} {event_text} older than {days_to_keep} {day_text}.",
+            )
+            return HttpResponseRedirect(
+                reverse("admin:django_github_app_eventlog_changelist")
+            )
+
+        context = {
+            **self.admin_site.each_context(request),
+            "form": form,
+            "opts": self.model._meta,
+        }
+
+        if form.is_valid():
+            context["title"] = f"Confirm {self.model._meta.verbose_name} deletion"
+            template = "cleanup_confirmation.html"
+        else:
+            context["title"] = f"Clean up {self.model._meta.verbose_name_plural}"
+            template = "cleanup.html"
+
+        return render(request, f"admin/django_github_app/eventlog/{template}", context)
+
 
 @admin.register(Installation)
 class InstallationModelAdmin(admin.ModelAdmin):

src/django_github_app/templates/admin/django_github_app/eventlog/change_list.html

@@ -0,0 +1,9 @@
+{% extends "admin/change_list.html" %}
+{% load i18n admin_urls %}
+{% block object-tools-items %}
+    <li>
+        {% url opts|admin_urlname:'cleanup' as cleanup_url %}
+        <a href="{{ cleanup_url }}">{% blocktranslate with verbose_name_plural=opts.verbose_name_plural %}Clean up {{ verbose_name_plural }}{% endblocktranslate %}</a>
+    </li>
+    {{ block.super }}
+{% endblock %}

src/django_github_app/templates/admin/django_github_app/eventlog/cleanup.html

@@ -0,0 +1,35 @@
+{% extends "admin/base_site.html" %}
+{% load i18n admin_urls %}
+{% block bodyclass %}
+    {{ block.super }} app-{{ opts.app_label }} model-{{ opts.model_name }} delete-confirmation
+{% endblock %}
+{% block breadcrumbs %}
+    {% include "admin/django_github_app/eventlog/includes/cleanup_breadcrumbs.html" %}
+{% endblock %}
+{% block content %}
+    <form method="post">
+        {% csrf_token %}
+        <div class="module">
+            {% for field in form %}
+                <div>
+                    {{ field.errors }}
+                    {{ field.label_tag }}
+                    {{ field }}
+                    <div class="help"
+                         {% if field.id_for_label %}id="{{ field.id_for_label }}_helptext"{% endif %}>
+                        <div>{{ field.help_text|safe }}</div>
+                    </div>
+                </div>
+            {% endfor %}
+        </div>
+        <div class="submit-row">
+            <input type="submit"
+                   value="{% translate 'Delete' %}"
+                   class="default"
+                   name="_save">
+            {% url opts|admin_urlname:'changelist' as changelist_url %}
+            <a href="{% add_preserved_filters changelist_url %}"
+               class="button cancel-link">{% translate 'Cancel' %}</a>
+        </div>
+    </form>
+{% endblock %}

src/django_github_app/templates/admin/django_github_app/eventlog/cleanup_confirmation.html

@@ -0,0 +1,28 @@
+{% extends "admin/delete_confirmation.html" %}
+{% load i18n admin_urls %}
+{% block breadcrumbs %}
+    {% include "admin/django_github_app/eventlog/includes/cleanup_breadcrumbs.html" %}
+{% endblock %}
+{% block delete_confirm %}
+    <p>
+        {% blocktranslate count counter=form.to_delete_count with verbose_name=opts.verbose_name verbose_name_plural=opts.verbose_name_plural days_to_keep=form.cleaned_data.days_to_keep %}You are about to delete {{ counter }} {{ verbose_name }} older than {{ days_to_keep }} days.{% plural %}You are about to delete {{ counter }} {{ verbose_name_plural }} older than {{ days_to_keep }} days.{% endblocktranslate %}
+    </p>
+    <p>
+        {% blocktranslate with verbose_name_plural=opts.verbose_name_plural cutoff_date=form.cutoff_date %}All {{ verbose_name_plural }} received before {{ cutoff_date }} will be permanently deleted.{% endblocktranslate %}
+    </p>
+    {% if form.to_delete_count %}
+        <h2>{% translate "Summary" %}</h2>
+        <ul>
+            <li>{% blocktranslate with name=opts.verbose_name_plural count=form.to_delete_count %}{{ name }}: {{ count }}{% endblocktranslate %}</li>
+        </ul>
+    {% endif %}
+    <form method="post">
+        {% csrf_token %}
+        <div>
+            <input type="hidden" name="post" value="yes">
+            <input type="hidden" name="days_to_keep" value="{{ form.cleaned_data.days_to_keep }}">
+            <input type="submit" value="{% translate "Yes, I'm sure" %}">
+            <a href="#" class="button cancel-link">{% translate 'No, take me back' %}</a>
+        </div>
+    </form>
+{% endblock %}

src/django_github_app/templates/admin/django_github_app/eventlog/includes/cleanup_breadcrumbs.html

@@ -0,0 +1,7 @@
+{% load i18n admin_urls %}
+<div class="breadcrumbs">
+    <a href="{% url 'admin:index' %}">{% translate 'Home' %}</a>
+    &rsaquo; <a href="{% url 'admin:app_list' app_label=opts.app_label %}">{{ opts.app_config.verbose_name }}</a>
+    &rsaquo; <a href="{% url opts|admin_urlname:'changelist' %}">{{ opts.verbose_name_plural|capfirst }}</a>
+    &rsaquo; {{ title }}
+</div>

tests/test_admin.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+import datetime
+from unittest.mock import patch
+
+import pytest
+from django.contrib.admin.sites import AdminSite
+from django.contrib.auth import get_user_model
+from django.contrib.messages import get_messages
+from django.test import RequestFactory
+from django.urls import reverse
+from django.utils import timezone
+
+from django_github_app.admin import EventLogModelAdmin
+from django_github_app.models import EventLog
+
+User = get_user_model()
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def admin_user():
+    return User.objects.create_superuser(
+        username="admin", email="admin@test.com", password="adminpass"
+    )
+
+
+@pytest.fixture
+def admin_site():
+    return AdminSite()
+
+
+@pytest.fixture
+def eventlog_admin(admin_site):
+    return EventLogModelAdmin(EventLog, admin_site)
+
+
+@pytest.fixture
+def factory():
+    return RequestFactory()
+
+
+class TestEventLogModelAdmin:
+    def test_cleanup_url_exists(self, client, admin_user):
+        client.login(username="admin", password="adminpass")
+        response = client.get(reverse("admin:django_github_app_eventlog_changelist"))
+
+        assert response.status_code == 200
+        # Check that the cleanup URL is in the rendered HTML
+        cleanup_url = reverse("admin:django_github_app_eventlog_cleanup")
+        assert cleanup_url.encode() in response.content
+
+    def test_cleanup_view_get(self, factory, admin_user, eventlog_admin):
+        request = factory.get("/admin/django_github_app/eventlog/cleanup/")
+        request.user = admin_user
+        response = eventlog_admin.cleanup_view(request)
+
+        assert response.status_code == 200
+        assert b"Clean up event logs" in response.content
+        assert b"Days to keep" in response.content
+
+    def test_cleanup_view_post_shows_confirmation(self, client, admin_user, baker):
+        # Create some test events
+        now = timezone.now()
+        baker.make(EventLog, _quantity=3, received_at=now - datetime.timedelta(days=10))
+        baker.make(EventLog, _quantity=2, received_at=now - datetime.timedelta(days=2))
+
+        client.login(username="admin", password="adminpass")
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"days_to_keep": "5"},
+        )
+
+        assert response.status_code == 200
+        assert b"You are about to delete 3 event logs" in response.content
+        assert b"Yes, I" in response.content and b"m sure" in response.content
+
+    @patch("django_github_app.models.EventLog.objects.cleanup_events")
+    def test_cleanup_view_confirm_deletion(self, mock_cleanup, client, admin_user):
+        mock_cleanup.return_value = (5, {"django_github_app.EventLog": 5})
+
+        client.login(username="admin", password="adminpass")
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"post": "yes", "days_to_keep": "3"},
+        )
+
+        assert response.status_code == 302
+        assert response.url == reverse("admin:django_github_app_eventlog_changelist")
+        mock_cleanup.assert_called_once_with(3)
+
+        # Check success message
+        messages = list(get_messages(response.wsgi_request))
+        assert len(messages) == 1
+        assert "Successfully deleted 5 events older than 3 days" in str(messages[0])
+
+    @patch("django_github_app.models.EventLog.objects.cleanup_events")
+    def test_cleanup_view_confirm_deletion_singular_day(
+        self, mock_cleanup, client, admin_user
+    ):
+        mock_cleanup.return_value = (2, {"django_github_app.EventLog": 2})
+
+        client.login(username="admin", password="adminpass")
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"post": "yes", "days_to_keep": "1"},
+        )
+
+        assert response.status_code == 302
+
+        # Check success message uses singular "day" and plural "events"
+        messages = list(get_messages(response.wsgi_request))
+        assert len(messages) == 1
+        assert "Successfully deleted 2 events older than 1 day" in str(messages[0])
+
+    @patch("django_github_app.models.EventLog.objects.cleanup_events")
+    def test_cleanup_view_confirm_deletion_zero_events(
+        self, mock_cleanup, client, admin_user
+    ):
+        mock_cleanup.return_value = (0, {})
+
+        client.login(username="admin", password="adminpass")
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"post": "yes", "days_to_keep": "7"},
+        )
+
+        assert response.status_code == 302
+
+        # Check success message uses plural "events" for zero
+        messages = list(get_messages(response.wsgi_request))
+        assert len(messages) == 1
+        assert "Successfully deleted 0 events older than 7 days" in str(messages[0])
+
+    def test_cleanup_view_integration(self, client, admin_user, baker):
+        now = timezone.now()
+
+        # Create test EventLog entries using baker
+        old_event = baker.make(
+            EventLog,
+            event="push",
+            payload={"action": "created"},
+            received_at=now - datetime.timedelta(days=10),
+        )
+        recent_event = baker.make(
+            EventLog,
+            event="pull_request",
+            payload={"action": "opened"},
+            received_at=now - datetime.timedelta(days=2),
+        )
+
+        client.login(username="admin", password="adminpass")
+
+        # Test GET request
+        response = client.get(reverse("admin:django_github_app_eventlog_cleanup"))
+        assert response.status_code == 200
+
+        # Test POST request - Step 1: Show confirmation
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"days_to_keep": "5"},
+        )
+        assert response.status_code == 200
+        assert b"You are about to delete 1 event log" in response.content
+
+        # Test POST request - Step 2: Confirm deletion
+        response = client.post(
+            reverse("admin:django_github_app_eventlog_cleanup"),
+            {"post": "yes", "days_to_keep": "5"},
+        )
+        assert response.status_code == 302
+
+        # Check that old event was deleted and recent event remains
+        assert not EventLog.objects.filter(id=old_event.id).exists()
+        assert EventLog.objects.filter(id=recent_event.id).exists()
+
+        # Check success message
+        messages = list(get_messages(response.wsgi_request))
+        assert len(messages) == 1
+        assert "Successfully deleted 1 event older than 5 days" in str(messages[0])