Notifications: Convert to classes

dojo/engagement/views.py

@@ -28,7 +28,6 @@
 from openpyxl.styles import Font
 
 import dojo.jira_link.helper as jira_helper
-import dojo.notifications.helper as notifications_helper
 import dojo.risk_acceptance.helper as ra_helper
 from dojo.authorization.authorization import user_has_permission_or_403
 from dojo.authorization.authorization_decorators import user_is_authorized
@@ -662,7 +661,15 @@ def add_tests(request, eid):
                 "Test added successfully.",
                 extra_tags="alert-success")
 
-            notifications_helper.notify_test_created(new_test)
+            create_notification(
+                event="test_added",
+                title=f"Test created for {new_test.engagement.product}: {new_test.engagement.name}: {new_test}",
+                test=new_test,
+                engagement=new_test.engagement,
+                product=new_test.engagement.product,
+                url=reverse("view_test", args=(new_test.id,)),
+                url_api=reverse("test-detail", args=(new_test.id,)),
+            )
 
             if "_Add Another Test" in request.POST:
                 return HttpResponseRedirect(

dojo/importers/base_importer.py

@@ -5,6 +5,7 @@
 from django.core.exceptions import ValidationError
 from django.core.files.base import ContentFile
 from django.core.files.uploadedfile import TemporaryUploadedFile
+from django.urls import reverse
 from django.utils.timezone import make_aware
 
 import dojo.finding.helper as finding_helper
@@ -28,6 +29,7 @@
     Test_Type,
     Vulnerability_Id,
 )
+from dojo.notifications.helper import create_notification
 from dojo.tools.factory import get_parser
 from dojo.utils import max_safe
 
@@ -719,3 +721,43 @@ def mitigate_finding(
             finding.save(dedupe_option=False)
         else:
             finding.save(dedupe_option=False, push_to_jira=self.push_to_jira)
+
+    def notify_scan_added(
+        self,
+        test,
+        updated_count,
+        new_findings=[],
+        findings_mitigated=[],
+        findings_reactivated=[],
+        findings_untouched=[],
+    ):
+        logger.debug("Scan added notifications")
+
+        new_findings = sorted(new_findings, key=lambda x: x.numerical_severity)
+        findings_mitigated = sorted(findings_mitigated, key=lambda x: x.numerical_severity)
+        findings_reactivated = sorted(findings_reactivated, key=lambda x: x.numerical_severity)
+        findings_untouched = sorted(findings_untouched, key=lambda x: x.numerical_severity)
+
+        title = (
+            f"Created/Updated {updated_count} findings for {test.engagement.product}: {test.engagement.name}: {test}"
+        )
+
+        if updated_count == 0:
+            event = "scan_added_empty"
+        else:
+            event = "scan_added"
+
+        create_notification(
+            event=event,
+            title=title,
+            findings_new=new_findings,
+            findings_mitigated=findings_mitigated,
+            findings_reactivated=findings_reactivated,
+            finding_count=updated_count,
+            test=test,
+            engagement=test.engagement,
+            product=test.engagement.product,
+            findings_untouched=findings_untouched,
+            url=reverse("view_test", args=(test.id,)),
+            url_api=reverse("test-detail", args=(test.id,)),
+        )

dojo/importers/default_importer.py

@@ -3,10 +3,10 @@
 from django.core.files.uploadedfile import TemporaryUploadedFile
 from django.core.serializers import deserialize, serialize
 from django.db.models.query_utils import Q
+from django.urls import reverse
 
 import dojo.finding.helper as finding_helper
 import dojo.jira_link.helper as jira_helper
-import dojo.notifications.helper as notifications_helper
 from dojo.importers.base_importer import BaseImporter, Parser
 from dojo.importers.options import ImporterOptions
 from dojo.models import (
@@ -15,6 +15,7 @@
     Test,
     Test_Import,
 )
+from dojo.notifications.helper import create_notification
 
 logger = logging.getLogger(__name__)
 deduplicationLogger = logging.getLogger("dojo.specific-loggers.deduplication")
@@ -126,9 +127,17 @@ def process_scan(
         )
         # Send out some notifications to the user
         logger.debug("IMPORT_SCAN: Generating notifications")
-        notifications_helper.notify_test_created(self.test)
+        create_notification(
+            event="test_added",
+            title=f"Test created for {self.test.engagement.product}: {self.test.engagement.name}: {self.test}",
+            test=self.test,
+            engagement=self.test.engagement,
+            product=self.test.engagement.product,
+            url=reverse("view_test", args=(self.test.id,)),
+            url_api=reverse("test-detail", args=(self.test.id,)),
+        )
         updated_count = len(new_findings) + len(closed_findings)
-        notifications_helper.notify_scan_added(
+        self.notify_scan_added(
             self.test,
             updated_count,
             new_findings=new_findings,

dojo/importers/default_reimporter.py

@@ -6,7 +6,6 @@
 
 import dojo.finding.helper as finding_helper
 import dojo.jira_link.helper as jira_helper
-import dojo.notifications.helper as notifications_helper
 from dojo.importers.base_importer import BaseImporter, Parser
 from dojo.importers.options import ImporterOptions
 from dojo.models import (
@@ -128,7 +127,7 @@ def process_scan(
         updated_count = (
             len(closed_findings) + len(reactivated_findings) + len(new_findings)
         )
-        notifications_helper.notify_scan_added(
+        self.notify_scan_added(
             self.test,
             updated_count,
             new_findings=new_findings,