events: add mark_all_seen

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

authentik/events/api/notification.py

@@ -1,8 +1,13 @@
 """Notification API Views"""
 from django_filters.rest_framework import DjangoFilterBackend
+from drf_spectacular.types import OpenApiTypes
+from drf_spectacular.utils import OpenApiResponse, extend_schema
 from rest_framework import mixins
+from rest_framework.decorators import action
 from rest_framework.fields import ReadOnlyField
 from rest_framework.filters import OrderingFilter, SearchFilter
+from rest_framework.request import Request
+from rest_framework.response import Response
 from rest_framework.serializers import ModelSerializer
 from rest_framework.viewsets import GenericViewSet
 
@@ -53,3 +58,18 @@ class NotificationViewSet(
     ]
     permission_classes = [OwnerPermissions]
     filter_backends = [OwnerFilter, DjangoFilterBackend, OrderingFilter, SearchFilter]
+
+    @extend_schema(
+        request=OpenApiTypes.NONE,
+        responses={
+            204: OpenApiResponse(description="Marked tasks as read successfully."),
+        },
+    )
+    @action(detail=False, methods=["post"])
+    def mark_all_seen(self, request: Request) -> Response:
+        """Mark all the user's notifications as seen"""
+        notifications = Notification.objects.filter(user=request.user)
+        for notification in notifications:
+            notification.seen = True
+        Notification.objects.bulk_update(notifications, ["seen"])
+        return Response({}, status=204)

authentik/events/tests/test_api.py

@@ -1,18 +1,19 @@
 """Event API tests"""
 
+from authentik.events.api.notification import NotificationSerializer
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
 from authentik.core.models import User
-from authentik.events.models import Event, EventAction
+from authentik.events.models import Event, EventAction, Notification, NotificationSeverity
 
 
 class TestEventsAPI(APITestCase):
     """Test Event API"""
 
     def setUp(self) -> None:
-        user = User.objects.get(username="akadmin")
-        self.client.force_login(user)
+        self.user = User.objects.get(username="akadmin")
+        self.client.force_login(self.user)
 
     def test_top_n(self):
         """Test top_per_user"""
@@ -30,3 +31,14 @@ class TestEventsAPI(APITestCase):
             reverse("authentik_api:event-actions"),
         )
         self.assertEqual(response.status_code, 200)
+
+    def test_notifications(self):
+        """Test notifications"""
+        notification = Notification.objects.create(
+            user=self.user, severity=NotificationSeverity.ALERT, body="", seen=False
+        )
+        self.client.post(
+            reverse("authentik_api:notification-mark-all-seen"),
+        )
+        notification.refresh_from_db()
+        self.assertTrue(notification.seen)