Fixes #11942: Add notifications model and REST API impl

netbox/extras/api/serializers.py

@@ -11,6 +11,7 @@ from dcim.api.nested_serializers import (
 from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup
 from extras.choices import *
 from extras.models import *
+from extras.models.staging import Notification
 from extras.utils import FeatureQuery
 from netbox.api.exceptions import SerializerNotFound
 from netbox.api.fields import ChoiceField, ContentTypeField, SerializedPKRelatedField
@@ -47,6 +48,7 @@ __all__ = (
     'ScriptSerializer',
     'TagSerializer',
     'WebhookSerializer',
+    'NotificationSerializer',
 )
 
 
@@ -525,3 +527,13 @@ class ContentTypeSerializer(BaseModelSerializer):
     class Meta:
         model = ContentType
         fields = ['id', 'url', 'display', 'app_label', 'model']
+
+
+class NotificationSerializer(BaseModelSerializer):
+    url = serializers.HyperlinkedIdentityField(view_name='extras-api:notifications-detail')
+
+    class Meta:
+        model = Notification
+        fields = [
+            'id', 'url', 'created', 'title', 'content', 'read'
+        ]

netbox/extras/api/urls.py

@@ -19,6 +19,7 @@ router.register('scripts', views.ScriptViewSet, basename='script')
 router.register('object-changes', views.ObjectChangeViewSet)
 router.register('job-results', views.JobResultViewSet)
 router.register('content-types', views.ContentTypeViewSet)
+router.register('notifications', views.NotificationViewSet, basename='notifications')
 
 app_name = 'extras-api'
 urlpatterns = router.urls

netbox/extras/api/views.py

@@ -1,6 +1,7 @@
 from django.contrib.contenttypes.models import ContentType
 from django.http import Http404
 from django_rq.queues import get_connection
+from django.shortcuts import get_object_or_404
 from rest_framework import status
 from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
@@ -14,6 +15,7 @@ from extras import filtersets
 from extras.choices import JobResultStatusChoices
 from extras.models import *
 from extras.models import CustomField
+from extras.models.staging import Notification
 from extras.reports import get_report, get_reports, run_report
 from extras.scripts import get_script, get_scripts, run_script
 from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
@@ -389,3 +391,31 @@ class ContentTypeViewSet(ReadOnlyModelViewSet):
     queryset = ContentType.objects.order_by('app_label', 'model')
     serializer_class = serializers.ContentTypeSerializer
     filterset_class = filtersets.ContentTypeFilterSet
+
+
+class NotificationViewSet(ReadOnlyModelViewSet):
+    """
+    Notification views API.
+
+    Note that this is not quite a read only viewset, we implement partial_update and delete.
+    The use of ReadOnlyModelViewSet is simply to reuse some existing code around serving the
+    list/retrieve endpoints and getting pagination for free.
+    """
+
+    permission_classes = [IsAuthenticatedOrLoginNotRequired]
+    serializer_class = serializers.NotificationSerializer
+
+    def get_queryset(self):
+        return Notification.objects.filter(user__id=self.request.user.id).order_by('created')
+
+    def partial_update(self, request, pk=None):
+        n = get_object_or_404(self.get_queryset(), pk=pk)
+        serializer = self.get_serializer(n, data=request.data, partial=True)
+        serializer.is_valid()
+        serializer.save()
+        return Response(serializer.data)
+
+    def destroy(self, _, pk=None):
+        n = get_object_or_404(self.get_queryset(), pk=pk)
+        n.delete()
+        return Response(status=status.HTTP_200_OK)

netbox/extras/models/staging.py

@@ -4,14 +4,17 @@ from django.contrib.auth import get_user_model
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.contenttypes.models import ContentType
 from django.db import models, transaction
+from django.conf import settings
+from django.urls import reverse
 
 from extras.choices import ChangeActionChoices
-from netbox.models import ChangeLoggedModel
+from netbox.models import ChangeLoggedModel, NetBoxModel
 from utilities.utils import deserialize_object
 
 __all__ = (
     'Branch',
     'StagedChange',
+    'Notification',
 )
 
 logger = logging.getLogger('netbox.staging')
@@ -112,3 +115,35 @@ class StagedChange(ChangeLoggedModel):
             instance = self.model.objects.get(pk=self.object_id)
             logger.info(f'Deleting {self.model._meta.verbose_name} {instance}')
             instance.delete()
+
+
+class Notification(NetBoxModel):
+    """
+    Notifications allow users to keep up to date with system events or requests.
+
+    Originally implemented to support the use case when a user is notified that a ReviewRequest
+    has been assigned to her.
+    """
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.CASCADE,
+    )
+
+    title = models.CharField(
+        max_length=256
+    )
+
+    content = models.TextField()
+
+    read = models.BooleanField(
+        default=False
+    )
+
+    def __str__(self):
+        return f'[UserID: {self.user.pk}, Read: {self.read}] {self.title} {self.content}'
+
+    def get_absolute_url(self):
+        return reverse('extras-api:notifications-detail', args=[self.pk])
+
+    class Meta:
+        ordering = ('pk',)

netbox/extras/tests/test_api.py

@@ -12,9 +12,11 @@ from rq import Worker
 from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, Location, RackRole, Site
 from extras.api.views import ReportViewSet, ScriptViewSet
 from extras.models import *
+from extras.models.staging import Notification
 from extras.reports import Report
 from extras.scripts import BooleanVar, IntegerVar, Script, StringVar
 from utilities.testing import APITestCase, APIViewTestCases
+from users.models import Token
 
 rq_worker_running = Worker.count(get_connection('default'))
 
@@ -699,3 +701,86 @@ class ContentTypeTest(APITestCase):
 
         url = reverse('extras-api:contenttype-detail', kwargs={'pk': contenttype.pk})
         self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK)
+
+
+class NotificationTest(APITestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.u1 = User.objects.create_user(username='user1')
+        cls.u2 = User.objects.create_user(username='user2')
+        cls.u1_key = Token.objects.create(user=cls.u1).key
+        cls.u2_key = Token.objects.create(user=cls.u2).key
+
+    def test_list(self):
+        test_notifs = [
+            {'user_id': self.u1, 'title': 'notif1', 'content': 'c1'},
+            {'user_id': self.u1, 'title': 'notif2', 'content': 'c2'},
+            {'user_id': self.u2, 'title': 'notif3', 'content': 'c3'},
+        ]
+        for t in test_notifs:
+            n = Notification(user=t['user_id'], title=t['title'], content=t['content'])
+            n.save()
+
+        u1_headers = {'HTTP_AUTHORIZATION': f'Token {self.u1_key}'}
+        response = self.client.get(reverse('extras-api:notifications-list'), **u1_headers)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        # Make sure we get only the notifications that belong to the authenticated user.
+        self.assertEqual(response.data['count'], Notification.objects.filter(user__id=self.u1.id).count())
+
+    def test_retrieve(self):
+        expected_title = 'expected title'
+        expected_content = 'some content'
+        n = Notification(user=self.u1, title=expected_title, content=expected_content)
+        n.save()
+        u1_headers = {'HTTP_AUTHORIZATION': f'Token {self.u1_key}'}
+        url = reverse('extras-api:notifications-detail', kwargs={'pk': n.id})
+        response = self.client.get(url, **u1_headers)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['title'], expected_title)
+        self.assertEqual(response.data['content'], expected_content)
+        self.assertEqual(response.data['read'], False)  # The default value is False
+
+        # Make sure user2 can't retrieve user1 notifications.
+        u2_headers = {'HTTP_AUTHORIZATION': f'Token {self.u2_key}'}
+        response = self.client.get(url, **u2_headers)
+        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+
+    def test_partial_update(self):
+        n = Notification(user=self.u1, title='t1', content='c1')
+        n.save()
+        expected_title = 'expected_title'
+        u1_headers = {'HTTP_AUTHORIZATION': f'Token {self.u1_key}'}
+        url = reverse('extras-api:notifications-detail', kwargs={'pk': n.id})
+        data = {
+            'title': expected_title,
+            'read': True
+        }
+        response = self.client.patch(url, data, format='json', **u1_headers)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['title'], expected_title)
+        self.assertEqual(response.data['read'], True)
+
+        # Make sure user2 can't update user1 notifications
+        u2_headers = {'HTTP_AUTHORIZATION': f'Token {self.u2_key}'}
+        response = self.client.patch(url, data, format='json', **u2_headers)
+        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+
+    def test_delete(self):
+        n = Notification(user=self.u1, title='t1', content='c1')
+        n.save()
+        url = reverse('extras-api:notifications-detail', kwargs={'pk': n.id})
+
+        # First make sure user2 can't delete user1 notifications
+        u2_headers = {'HTTP_AUTHORIZATION': f'Token {self.u2_key}'}
+        response = self.client.delete(url, **u2_headers)
+        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
+
+        u1_headers = {'HTTP_AUTHORIZATION': f'Token {self.u1_key}'}
+        response = self.client.delete(url, **u1_headers)
+        self.assertHttpStatus(response, status.HTTP_200_OK)  # Assert deletion
+
+        # Validate that it's actually gone.
+        response = self.client.get(url, **u1_headers)
+        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

netbox/extras/urls.py

@@ -1,6 +1,7 @@
 from django.urls import include, path, re_path
 
 from extras import views
+from extras.api.views import NotificationViewSet
 from utilities.urls import get_model_urls