Add notification when finding is created via the API (#13732)

* Add notification when finding is created via the API

* fix: add tests

* fix: Ruff linter COM812 errors

* fix: move tests to test_notifications.TestNotificationTriggersApi

* fix: W293 Blank line contains whitespace

---------

Co-authored-by: Julien Reitzel <[email protected]>

Author: Juu

dojo/api_v2/serializers.py

@@ -114,6 +114,7 @@
     Vulnerability_Id_Template,
     get_current_date,
 )
+from dojo.notifications.helper import create_notification
 from dojo.product_announcements import (
     LargeScanSizeProductAnnouncement,
     ScanTypeProductAnnouncement,
@@ -1949,6 +1950,16 @@ def create(self, validated_data):
         if push_to_jira:
             jira_helper.push_to_jira(new_finding)
 
+        # Create a notification
+        create_notification(
+            event="finding_added",
+            title=_("Addition of %s") % new_finding.title,
+            finding=new_finding,
+            description=_('Finding "%s" was added by %s') % (new_finding.title, new_finding.reporter),
+            url=reverse("view_finding", args=(new_finding.id,)),
+            icon="exclamation-triangle",
+        )
+
         return new_finding
 
     def validate(self, data):

dojo/test/views.py

@@ -607,7 +607,6 @@ def process_forms(self, request: HttpRequest, test: Test, context: dict):
 
             # Note: this notification has not be moved to "@receiver(post_save, sender=Finding)" method as many other notifications
             # Because it could generate too much noise, we keep it here only for findings created by hand in WebUI
-            # TODO: but same should be implemented for API endpoint
 
             # Create a notification
             create_notification(

unittests/test_notifications.py

@@ -405,13 +405,122 @@ def setUp(self):
         token = Token.objects.get(user__username="admin")
         self.client = APIClient()
         self.client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+        self.admin = User.objects.get(username="admin")
+        self.base_url = "/api/v2/findings/"
+
+    def _minimal_create_payload(self, title: str):
+        return {
+            "test": 3,
+            "found_by": [],
+            "title": title,
+            "date": "2020-05-20",
+            "cwe": 1,
+            "severity": "High",
+            "description": "TEST finding for notification",
+            "mitigation": "MITIGATION",
+            "impact": "HIGH",
+            "references": "",
+            "active": True,
+            "verified": False,
+            "false_p": False,
+            "duplicate": False,
+            "out_of_scope": False,
+            "under_review": False,
+            "under_defect_review": False,
+            "numerical_severity": "S0",
+        }
 
     @patch("dojo.notifications.helper.NotificationManager._process_notifications")
     def test_auditlog_on(self, mock):
         prod_type = Product_Type.objects.create(name="notif prod type API")
         self.client.delete(reverse("product_type-detail", args=(prod_type.pk,)), format="json")
         self.assertEqual(mock.call_args_list[-1].kwargs["description"], 'The product type "notif prod type API" was deleted by admin')
 
+    @patch("dojo.api_v2.serializers.create_notification")
+    def test_create_calls_notification_with_auto_assigned_reporter(self, mock_create_notification):
+        """Test that create_notification is called when creating a finding without explicit reporter."""
+        payload = self._minimal_create_payload("Finding with auto-assigned reporter notification")
+
+        response = self.client.post(self.base_url, payload, format="json")
+        self.assertEqual(201, response.status_code, response.content[:1000])
+
+        # Verify notification was called
+        mock_create_notification.assert_called_once()
+        call_args = mock_create_notification.call_args
+
+        # Check the notification parameters
+        self.assertEqual(call_args[1]["event"], "finding_added")
+        self.assertEqual(call_args[1]["title"], "Addition of Finding With Auto-Assigned Reporter Notification")
+        self.assertEqual(
+            call_args[1]["description"],
+            f'Finding "Finding With Auto-Assigned Reporter Notification" was added by {self.admin}',
+        )
+        self.assertEqual(call_args[1]["icon"], "exclamation-triangle")
+
+        # Verify the finding was created successfully
+        created_id = response.data.get("id")
+        self.assertIsNotNone(created_id)
+        created_finding = Finding.objects.get(id=created_id)
+        self.assertEqual(created_finding.reporter, self.admin)
+
+    @patch("dojo.api_v2.serializers.create_notification")
+    def test_create_calls_notification_with_explicit_reporter(self, mock_create_notification):
+        """Test that create_notification is called when creating a finding with explicit reporter."""
+        # Create another user to use as explicit reporter
+        explicit_reporter = User.objects.create(username="explicit_reporter", email="[email protected]")
+
+        payload = self._minimal_create_payload("Finding with explicit reporter notification")
+        payload["reporter"] = explicit_reporter.id
+
+        response = self.client.post(self.base_url, payload, format="json")
+        self.assertEqual(201, response.status_code, response.content[:1000])
+
+        # Verify notification was called
+        mock_create_notification.assert_called_once()
+        call_args = mock_create_notification.call_args
+
+        # Check the notification parameters
+        self.assertEqual(call_args[1]["event"], "finding_added")
+        self.assertEqual(call_args[1]["title"], "Addition of Finding With Explicit Reporter Notification")
+        self.assertEqual(
+            call_args[1]["description"],
+            f'Finding "Finding With Explicit Reporter Notification" was added by {explicit_reporter}',
+        )
+        self.assertEqual(call_args[1]["icon"], "exclamation-triangle")
+
+        # Verify the finding was created with explicit reporter
+        created_id = response.data.get("id")
+        self.assertIsNotNone(created_id)
+        created_finding = Finding.objects.get(id=created_id)
+        self.assertEqual(created_finding.reporter, explicit_reporter)
+
+    @patch("dojo.api_v2.serializers.create_notification")
+    def test_notification_parameters_are_correct(self, mock_create_notification):
+        """Test that all notification parameters are properly formatted and passed."""
+        payload = self._minimal_create_payload("Test Finding for Parameter Validation")
+
+        response = self.client.post(self.base_url, payload, format="json")
+        self.assertEqual(201, response.status_code, response.content[:1000])
+
+        # Get the created finding to verify URL formation
+        created_id = response.data.get("id")
+        created_finding = Finding.objects.get(id=created_id)
+
+        # Verify notification was called with correct parameters
+        mock_create_notification.assert_called_once()
+        call_args = mock_create_notification.call_args
+
+        # Verify all required parameters exist
+        self.assertEqual(call_args[1]["event"], "finding_added")
+        self.assertEqual(call_args[1]["title"], "Addition of Test Finding for Parameter Validation")
+        self.assertEqual(
+            call_args[1]["description"],
+            f'Finding "Test Finding for Parameter Validation" was added by {self.admin}',
+        )
+        self.assertEqual(call_args[1]["url"], f"/finding/{created_finding.id}")
+        self.assertEqual(call_args[1]["icon"], "exclamation-triangle")
+        self.assertEqual(call_args[1]["finding"], created_finding)
+
 
 class TestNotificationWebhooks(DojoTestCase):
     fixtures = ["dojo_testdata.json"]