Notifications: Clean up duplicate system notification entries

dojo/db_migrations/max_migration.txt

@@ -1 +1 @@
-0261_remove_url_insert_insert_remove_url_update_update_and_more
+0261_remove_url_insert_insert_remove_url_update_update_and_more

dojo/notifications/helper.py

@@ -118,9 +118,25 @@ def __init__(
     def _get_notifications_object(self) -> Notifications:
         """Set the system Notifications object on the class."""
         try:
-            return Notifications.objects.get(user=None, template=False)
-        except Exception:
-            return Notifications()
+            notifications, _ = Notifications.objects.get_or_create(
+                user=None, product=None, template=False,
+            )
+        except Notifications.MultipleObjectsReturned:
+            notifications = Notifications.objects.filter(
+                user=None,
+                product=None,
+                template=False,
+            ).first()
+            logger.warning(
+                "Multiple system notifications objects found, using the first one with id %s. Cleaning up the duplicate...",
+                notifications.id,
+            )
+            Notifications.objects.filter(
+                user=None,
+                product=None,
+                template=False,
+            ).exclude(id=notifications.id).delete()
+        return notifications
 
     def _get_system_settings(self) -> System_Settings:
         """Set the system settings object in the class."""

dojo/notifications/views.py

@@ -3,6 +3,7 @@
 import requests
 from django.contrib import messages
 from django.core.exceptions import PermissionDenied
+from django.db import transaction
 from django.http import Http404, HttpRequest, HttpResponseRedirect
 from django.shortcuts import get_object_or_404, render
 from django.urls import reverse
@@ -19,11 +20,10 @@
 
 class SystemNotificationsView(View):
     def get_notifications(self, request: HttpRequest):
-        try:
-            notifications = Notifications.objects.get(user=None, product__isnull=True, template=False)
-        except Notifications.DoesNotExist:
-            notifications = Notifications(user=None, template=False)
-
+        with transaction.atomic():
+            notifications, _ = Notifications.objects.get_or_create(
+                user=None, product=None, template=False,
+            )
         return notifications
 
     def check_user_permissions(self, request: HttpRequest):
@@ -97,10 +97,10 @@ def post(self, request: HttpRequest):
 
 class PersonalNotificationsView(SystemNotificationsView):
     def get_notifications(self, request: HttpRequest):
-        try:
-            notifications = Notifications.objects.get(user=request.user, product__isnull=True)
-        except Notifications.DoesNotExist:
-            notifications = Notifications(user=request.user)
+        with transaction.atomic():
+            notifications, _ = Notifications.objects.get_or_create(
+                user=request.user, product=None, template=False,
+            )
         return notifications
 
     def check_user_permissions(self, request: HttpRequest):
@@ -116,10 +116,10 @@ def set_breadcrumbs(self, request: HttpRequest):
 
 class TemplateNotificationsView(SystemNotificationsView):
     def get_notifications(self, request: HttpRequest):
-        try:
-            notifications = Notifications.objects.get(template=True)
-        except Notifications.DoesNotExist:
-            notifications = Notifications(user=None, template=True)
+        with transaction.atomic():
+            notifications, _ = Notifications.objects.get_or_create(
+                user=None, product=None, template=True,
+            )
         return notifications
 
     def get_scope(self):

unittests/test_notifications.py

@@ -578,23 +578,23 @@ def test_missing_system_webhook(self):
         with self.assertLogs("dojo.notifications.helper", level="INFO") as cm:
             manager = WebhookNotificationManger()
             manager.send_webhooks_notification(event="dummy")
-        self.assertIn("URLs for Webhooks not configured: skipping system notification", cm.output[0])
+        self.assertIn("URLs for Webhooks not configured: skipping system notification", cm.output[1])
 
     def test_missing_personal_webhook(self):
         # test data contains 2 entries but we need to test missing definition
         Notification_Webhooks.objects.all().delete()
         with self.assertLogs("dojo.notifications.helper", level="INFO") as cm:
             manager = WebhookNotificationManger()
             manager.send_webhooks_notification(event="dummy", user=Dojo_User.objects.get(username="admin"))
-        self.assertIn("URLs for Webhooks not configured for user '(admin)': skipping user notification", cm.output[0])
+        self.assertIn("URLs for Webhooks not configured for user '(admin)': skipping user notification", cm.output[1])
 
     def test_system_webhook_inactive(self):
         self.sys_wh.status = Notification_Webhooks.Status.STATUS_INACTIVE_PERMANENT
         self.sys_wh.save()
         with self.assertLogs("dojo.notifications.helper", level="INFO") as cm:
             manager = WebhookNotificationManger()
             manager.send_webhooks_notification(event="dummy")
-        self.assertIn("URL for Webhook 'My webhook endpoint' is not active: Permanently inactive (inactive_permanent)", cm.output[0])
+        self.assertIn("URL for Webhook 'My webhook endpoint' is not active: Permanently inactive (inactive_permanent)", cm.output[1])
 
     def test_system_webhook_sucessful(self):
         with self.assertLogs("dojo.notifications.helper", level="DEBUG") as cm: