Fix (notifications): Adds a method for defining/removing a notificati…

…on/token. Adds tests for the api and some fixes with them

src/api/v1/push_notifications/schemas.py

@@ -51,3 +51,7 @@ class PushNotificationReadManyInputSchema(serializers.Serializer):
 
 class PushNotificationReadManyOutputSchema(serializers.Serializer):
     read = serializers.IntegerField()
+
+
+class PushNotificationSetTokenInputSchema(serializers.Serializer):
+    notification_token = serializers.CharField()

src/api/v1/push_notifications/urls.py

@@ -3,6 +3,6 @@
 from . import views
 
 router = SimpleRouter(trailing_slash=False)
-router.register("notifications", views.PushNotificationViewSet, basename="push_notifications")
+router.register("", views.PushNotificationViewSet, basename="notifications")
 
 urlpatterns = router.urls

src/api/v1/push_notifications/views.py

@@ -4,9 +4,10 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
+from app.consts.http import HttpStatusCode
 from app.drf.openapi import limit_offset_openapi_schema, openapi_schema
 from app.drf.viewsets import AppViewSet
-from push_notifications import selectors, services
+from push_notifications import exc, selectors, services
 
 from . import schemas
 
@@ -35,7 +36,7 @@ def list(self, request: Request) -> Response:
         summary="Read many notifications",
         description="Reads a list of notifications",
         request=schemas.PushNotificationReadManyInputSchema,
-        responses={200: schemas.PushNotificationReadManyOutputSchema},
+        responses={HttpStatusCode.HTTP_200_OK: schemas.PushNotificationReadManyOutputSchema},
         tags=["push notifications"],
         operation_id="push-notifications-read",
         add_bad_request_response=True,
@@ -51,19 +52,50 @@ def read_many(self, request: Request) -> Response:
         summary="Read notification",
         description="Reads a single notification",
         request=None,
-        responses={200: schemas.PushNotificationOutputSchema},
+        responses={HttpStatusCode.HTTP_200_OK: schemas.PushNotificationOutputSchema},
         tags=["push notifications"],
         operation_id="push-notification-read",
         add_not_found_response=True,
         add_bad_request_response=True,
         add_unauthorized_response=True,
     )
     @action(methods=["PATCH"], detail=True)
-    def read(self, request: Request, id: int) -> Response:
+    def read(self, request: Request, pk: int) -> Response:
         notification = get_object_or_404(
             selectors.push_notification_get_viewable_qs(user=request.user),
-            pk=id,
+            pk=pk,
         )
         notification = services.push_notification_read(push_notification=notification)
         out_srlzr = schemas.PushNotificationOutputSchema(instance=notification)
         return Response(data=out_srlzr.data)
+
+    @openapi_schema(
+        summary="Set Notification token",
+        description="Defines the notification for the current user",
+        request=schemas.PushNotificationSetTokenInputSchema,
+        responses={HttpStatusCode.HTTP_200_OK: None},
+        tags=["push notifications"],
+        operation_id="push-notification-set-token",
+        add_unauthorized_response=True,
+        add_bad_request_response=True,
+        raises=[exc.InvalidNotificationToken],
+    )
+    @action(methods=["PUT"], detail=False)
+    def token(self, request: Request) -> Response:
+        data = self.get_valid_data(schemas.PushNotificationSetTokenInputSchema)
+        services.push_notification_set_token(user=request.user, **data)
+        return Response(status=HttpStatusCode.HTTP_200_OK)
+
+    @openapi_schema(
+        summary="Delete Notification token",
+        description="Removes the notification token from the current user (Opt-out)",
+        request=None,
+        responses={HttpStatusCode.HTTP_200_OK: None},
+        tags=["push notifications"],
+        operation_id="push-notification-delete-token",
+        add_unauthorized_response=True,
+    )
+    @token.mapping.delete
+    def delete_token(self, request: Request) -> Response:
+        services.push_notification_set_token(user=request.user, notification_token=None)
+        return Response(status=HttpStatusCode.HTTP_200_OK)

src/push_notifications/exc.py

@@ -1,5 +1,11 @@
+from django.utils.translation import gettext_lazy as _
+
 from app.exceptions import ApplicationError
 
 
 class UnableToSendPushNotification(ApplicationError):
     pass
+
+
+class InvalidNotificationToken(ApplicationError):
+    error_message = _("Invalid notification token, {reason}")

src/push_notifications/models.py

@@ -115,8 +115,8 @@ def __str__(self):
     def enriched_data(self):
         return {
             "id": self.id,
-            "createdAt": str(self.created_at),
-            "readAt": str(self.read_at),
+            "createdAt": self.created_at.isoformat(),
+            "readAt": self.read_at.isoformat() if self.read_at else None,
             "timeSinceCreated": timesince(self.created_at),
             "kind": self.kind,
             "meta": self.data,

src/push_notifications/services.py

@@ -1,7 +1,9 @@
 from datetime import datetime, timedelta
 from typing import Any, Sequence
 
+from django.core.exceptions import ValidationError
 from django.utils import timezone
+from django.utils.translation import gettext as _
 
 from app import consts
 from app.ext import di
@@ -14,7 +16,7 @@
 from app.models import BaseModel
 from users.models import User
 
-from . import models, selectors
+from . import exc, models, selectors
 
 
 def push_notification_create(
@@ -239,3 +241,21 @@ def push_notification_read_many(*, reader: User, ids: Sequence[int]) -> int:
     if ids:
         qs = qs.filter(pk__in=ids)
     return qs.update(read_at=timezone.now(), status=consts.push_notification.Status.READ)
+
+
+def push_notification_set_token(*, user: User, notification_token: str | None) -> None:
+    # We aren't calling full_clean here on User so we don't raise any other errors that
+    # aren't from this specific field
+
+    if notification_token is not None:
+        if not notification_token:
+            # Because we have blank=True, and we accept None values, raise if someone attempt
+            # to send a blank string ""
+            raise exc.InvalidNotificationToken(
+                message_format_kwargs={"reason": _("it may not be blank")}
+            )
+        max_length = User._meta.get_field("notification_token").max_length or 64
+        if len(notification_token) > max_length:
+            raise exc.InvalidNotificationToken(message_format_kwargs={"reason": _("value too big")})
+    user.notification_token = notification_token
+    user.save()

tests/conftest.py

@@ -1,4 +1,5 @@
 import pytest
+from django.utils import timezone
 
 from tests.fakes import FakePushNotificationExternalService
 from users.models import User
@@ -22,7 +23,8 @@ def visitor_user():
     user.set_password("password")
     user.full_clean()
     user.save()
-    return user
+    timezone.activate(user.time_zone)
+    yield user
 
 
 @pytest.fixture(autouse=True)