refactor: consolidate notifications into dojo/notifications/ package

dojo/api_v2/serializers.py

@@ -21,7 +21,7 @@
 from rest_framework import serializers
 from rest_framework.exceptions import NotFound
 from rest_framework.exceptions import ValidationError as RestFrameworkValidationError
-from rest_framework.fields import DictField, MultipleChoiceField
+from rest_framework.fields import DictField
 
 import dojo.finding.helper as finding_helper
 import dojo.risk_acceptance.helper as ra_helper
@@ -43,9 +43,7 @@
 from dojo.jira import services as jira_services
 from dojo.location.models import Location, LocationFindingReference
 from dojo.models import (
-    DEFAULT_NOTIFICATION,
     IMPORT_ACTIONS,
-    NOTIFICATION_CHOICES,
     SEVERITIES,
     SEVERITY_CHOICES,
     STATS_FIELDS,
@@ -82,8 +80,6 @@
     Note_Type,
     NoteHistory,
     Notes,
-    Notification_Webhooks,
-    Notifications,
     Product,
     Product_API_Scan_Configuration,
     Product_Group,
@@ -3069,110 +3065,7 @@ class FindingNoteSerializer(serializers.Serializer):
     note_id = serializers.IntegerField()
 
 
-class NotificationsSerializer(serializers.ModelSerializer):
-    product = serializers.PrimaryKeyRelatedField(
-        queryset=Product.objects.all(),
-        required=False,
-        default=None,
-        allow_null=True,
-    )
-    user = serializers.PrimaryKeyRelatedField(
-        queryset=Dojo_User.objects.all(),
-        required=False,
-        default=None,
-        allow_null=True,
-    )
-    product_type_added = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    product_added = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    engagement_added = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    test_added = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    scan_added = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    jira_update = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    upcoming_engagement = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    stale_engagement = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    auto_close_engagement = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    close_engagement = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    user_mentioned = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    code_review = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    review_requested = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    other = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    sla_breach = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    sla_breach_combined = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    risk_acceptance_expiration = MultipleChoiceField(
-        choices=NOTIFICATION_CHOICES, default=DEFAULT_NOTIFICATION,
-    )
-    template = serializers.BooleanField(default=False)
-
-    class Meta:
-        model = Notifications
-        fields = "__all__"
-
-    def validate(self, data):
-        user = None
-        product = None
-        template = False
-
-        if self.instance is not None:
-            user = self.instance.user
-            product = self.instance.product
-
-        if "user" in data:
-            user = data.get("user")
-        if "product" in data:
-            product = data.get("product")
-        if "template" in data:
-            template = data.get("template")
-
-        if (
-            template
-            and Notifications.objects.filter(template=True).count() > 0
-        ):
-            msg = "Notification template already exists"
-            raise ValidationError(msg)
-        if (
-            self.instance is None
-            or user != self.instance.user
-            or product != self.instance.product
-        ):
-            notifications = Notifications.objects.filter(
-                user=user, product=product, template=template,
-            ).count()
-            if notifications > 0:
-                msg = "Notification for user and product already exists"
-                raise ValidationError(msg)
-        return data
+from dojo.notifications.api.serializer import NotificationsSerializer  # noqa: E402, F401  -- backward compat
 
 
 class EngagementPresetsSerializer(serializers.ModelSerializer):
@@ -3349,7 +3242,4 @@ def create(self, validated_data):
             raise
 
 
-class NotificationWebhooksSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = Notification_Webhooks
-        fields = "__all__"
+from dojo.notifications.api.serializer import NotificationWebhooksSerializer  # noqa: E402, F401  -- backward compat

dojo/api_v2/views.py

@@ -119,8 +119,6 @@
     Note_Type,
     NoteHistory,
     Notes,
-    Notification_Webhooks,
-    Notifications,
     Product,
     Product_API_Scan_Configuration,
     Product_Group,
@@ -3406,21 +3404,6 @@ def queue_task_purge(self, request):
         return Response({"purged": purged})
 
 
-# Authorization: superuser
-@extend_schema_view(**schema_with_prefetch())
-class NotificationsViewSet(
-    PrefetchDojoModelViewSet,
-):
-    serializer_class = serializers.NotificationsSerializer
-    queryset = Notifications.objects.none()
-    filter_backends = (DjangoFilterBackend,)
-    filterset_fields = ["id", "user", "product", "template"]
-    permission_classes = (permissions.IsSuperUser, DjangoModelPermissions)
-
-    def get_queryset(self):
-        return Notifications.objects.all().order_by("id")
-
-
 @extend_schema_view(**schema_with_prefetch())
 class EngagementPresetsViewset(
     PrefetchDojoModelViewSet,
@@ -3683,13 +3666,3 @@ class AnnouncementViewSet(
 
     def get_queryset(self):
         return Announcement.objects.all().order_by("id")
-
-
-class NotificationWebhooksViewSet(
-    PrefetchDojoModelViewSet,
-):
-    serializer_class = serializers.NotificationWebhooksSerializer
-    queryset = Notification_Webhooks.objects.all()
-    filter_backends = (DjangoFilterBackend,)
-    filterset_fields = "__all__"
-    permission_classes = (permissions.IsSuperUser, DjangoModelPermissions)  # TODO: add permission also for other users

dojo/apps.py

@@ -84,6 +84,8 @@ def ready(self):
         import dojo.file_uploads.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
         import dojo.finding_group.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
         import dojo.notes.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
+        import dojo.notifications.admin  # noqa: PLC0415, F401 raised: AppRegistryNotReady
+        import dojo.notifications.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
         import dojo.product.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
         import dojo.product_type.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady
         import dojo.risk_acceptance.signals  # noqa: PLC0415, F401 raised: AppRegistryNotReady

dojo/context_processors.py

@@ -1,13 +1,12 @@
 import contextlib
-import time
 
 # import the settings file
 from django.conf import settings
 from django.contrib import messages
 
 from dojo.announcement.os_message import get_os_banner
 from dojo.labels import get_labels
-from dojo.models import Alerts, System_Settings, UserAnnouncement
+from dojo.models import System_Settings, UserAnnouncement
 
 
 def globalize_vars(request):
@@ -86,14 +85,6 @@ def bind_system_settings(request):
     return {"system_settings": system_settings}
 
 
-def bind_alert_count(request):
-    if not settings.DISABLE_ALERT_COUNTER:
-
-        if hasattr(request, "user") and request.user.is_authenticated:
-            return {"alert_count": Alerts.objects.filter(user_id=request.user).count()}
-    return {}
-
-
 def bind_announcement(request):
     with contextlib.suppress(Exception):  # TODO: this should be replaced with more meaningful exception
         if request.user.is_authenticated:
@@ -104,21 +95,10 @@ def bind_announcement(request):
     return {}
 
 
-def session_expiry_notification(request):
-    try:
-        if request.user.is_authenticated:
-            last_activity = request.session.get("_last_activity", time.time())
-            expiry_time = last_activity + settings.SESSION_COOKIE_AGE  # When the session will expire
-            warning_time = settings.SESSION_EXPIRE_WARNING  # Show warning X seconds before expiry
-            notify_time = expiry_time - warning_time
-        else:
-            notify_time = None
-    except Exception:
-        return {}
-    else:
-        return {
-            "session_notify_time": notify_time,
-        }
+from dojo.notifications.context_processors import (  # noqa: E402, F401  -- backward compat
+    bind_alert_count,
+    session_expiry_notification,
+)
 
 
 def labels(request):

dojo/forms.py

@@ -85,8 +85,6 @@
     Global_Role,
     Note_Type,
     Notes,
-    Notification_Webhooks,
-    Notifications,
     Objects_Product,
     Product,
     Product_API_Scan_Configuration,
@@ -3155,55 +3153,12 @@ class Meta:
         exclude = [""]
 
 
-class NotificationsForm(forms.ModelForm):
-
-    class Meta:
-        model = Notifications
-        exclude = ["template"]
-
-
-class NotificationsWebhookForm(forms.ModelForm):
-    class Meta:
-        model = Notification_Webhooks
-        exclude = []
-
-    def __init__(self, *args, **kwargs):
-        is_superuser = kwargs.pop("is_superuser", False)
-        super().__init__(*args, **kwargs)
-        if not is_superuser:  # Only superadmins can edit owner
-            self.fields["owner"].disabled = True  # TODO: needs to be tested
-
-
-class DeleteNotificationsWebhookForm(forms.ModelForm):
-    id = forms.IntegerField(required=True,
-                            widget=forms.widgets.HiddenInput())
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.fields["name"].disabled = True
-        self.fields["url"].disabled = True
-
-    class Meta:
-        model = Notification_Webhooks
-        fields = ["id", "name", "url"]
-
-
-class ProductNotificationsForm(forms.ModelForm):
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        if not self.instance.id:
-            self.initial["engagement_added"] = ""
-            self.initial["close_engagement"] = ""
-            self.initial["test_added"] = ""
-            self.initial["scan_added"] = ""
-            self.initial["sla_breach"] = ""
-            self.initial["sla_breach_combined"] = ""
-            self.initial["risk_acceptance_expiration"] = ""
-
-    class Meta:
-        model = Notifications
-        fields = ["engagement_added", "close_engagement", "test_added", "scan_added", "sla_breach", "sla_breach_combined", "risk_acceptance_expiration"]
+from dojo.notifications.ui.forms import (  # noqa: E402, F401  -- backward compat
+    DeleteNotificationsWebhookForm,
+    NotificationsForm,
+    NotificationsWebhookForm,
+    ProductNotificationsForm,
+)
 
 
 class AjaxChoiceField(forms.ChoiceField):