Consolidate notification code into dojo/notifications/ package

Move notification models, admin, services (helper.py), signals, tasks,
settings, context processors, forms, UI views/urls/templates, and API
serializers/viewsets/urls into dojo/notifications/, matching the
canonical dojo/url/ shape from CLAUDE.md.

- Notifications, Notification_Webhooks, Alerts (and constants) extracted
  from dojo/models.py into dojo/notifications/models.py; admin moved to
  dojo/notifications/admin.py using @admin.register(); registered from
  dojo/apps.py ready() to keep model class definition out of settings load.
- helper.py absorbs process_tag_notifications and sla_compute_and_notify
  from dojo/utils.py; all @app.task wrappers (log_generic_alert,
  add_alerts, cleanup_alerts, async_sla_compute_and_notify_task,
  send_{slack,msteams,mail,webhooks}_notification, webhook_reactivation,
  webhook_status_cleanup) consolidated into dojo/notifications/tasks.py.
- Default-Notifications-on-User-create logic moved out of dojo/utils.py
  into dojo/notifications/signals.py and registered via apps.py ready().
- 7 UI .html templates and 62 channel .tpl templates moved under
  dojo/notifications/templates/notifications/. New TEMPLATES DIRS entry
  added so notifications/{channel}/{event}.tpl lookups still resolve;
  view template_name strings updated where the namespace changed.
- Env-var schema extracted into dojo/notifications/settings.py
  (NOTIFICATIONS_ENV_DEFAULTS dict + populate_settings); settings.dist.py
  unpacks it into env() and applies values to globals().
- bind_alert_count and session_expiry_notification moved to
  dojo/notifications/context_processors.py; TEMPLATES paths updated.
- API NotificationsSerializer/NotificationWebhooksSerializer and
  Viewsets relocated under dojo/notifications/api/, exposed via
  add_notifications_urls(router) (CLAUDE.md API pattern).
- Celery beat task paths updated to dojo.notifications.tasks.*.
- Backward-compatible re-exports preserved at every original location
  (dojo/models.py, dojo/forms.py, dojo/utils.py, dojo/tasks.py,
  dojo/context_processors.py, dojo/api_v2/{serializers,views}.py,
  dojo/notifications/helper.py).
- No new migrations (string FK references are migration-equivalent).

Verified: manage.py check, makemigrations --check (no diff),
spectacular --fail-on-warn, and the full
test_notifications + test_apiv2_notifications + test_jira_webhook +
test_cleanup_alerts + test_rest_framework suite (1325 tests) pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Maffooch

dojo/api_v2/serializers.py

@@ -3138,110 +3138,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):
@@ -3418,7 +3315,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

@@ -3473,21 +3473,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,
@@ -3752,11 +3737,3 @@ 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

@@ -86,14 +86,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 +96,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

@@ -3240,55 +3240,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):