diff --git a/dojo/db_migrations/max_migration.txt b/dojo/db_migrations/max_migration.txt index 5b015714f08..d49ec6b6fcb 100644 --- a/dojo/db_migrations/max_migration.txt +++ b/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 \ No newline at end of file diff --git a/dojo/notifications/helper.py b/dojo/notifications/helper.py index 7318bae91af..610993cd3d7 100644 --- a/dojo/notifications/helper.py +++ b/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.""" diff --git a/dojo/notifications/views.py b/dojo/notifications/views.py index 3d96c4ab5e5..323a8b0ccf2 100644 --- a/dojo/notifications/views.py +++ b/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): diff --git a/unittests/test_notifications.py b/unittests/test_notifications.py index 2e897702fc4..065f488847b 100644 --- a/unittests/test_notifications.py +++ b/unittests/test_notifications.py @@ -578,7 +578,7 @@ 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 @@ -586,7 +586,7 @@ def test_missing_personal_webhook(self): 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 @@ -594,7 +594,7 @@ def test_system_webhook_inactive(self): 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: