fix tasks bug and add unit tests

Signed-off-by: tdruez <tdruez@aboutcode.org>

Author: tdruez

notification/tasks.py

@@ -12,6 +12,7 @@
 
 from django_rq import job
 
+from dje.models import get_unsecured_manager
 from notification.models import WebhookSubscription
 
 logger = logging.getLogger("dje")
@@ -36,7 +37,7 @@ def deliver_webhook_task(
     if instance_app_label and instance_model_name and instance_pk:
         try:
             model_class = apps.get_model(instance_app_label, instance_model_name)
-            instance = model_class.objects.get(pk=instance_pk)
+            instance = get_unsecured_manager(model_class).get(pk=instance_pk)
         except Exception:
             logger.error(
                 f"Instance {instance_app_label}.{instance_model_name} pk={instance_pk} not found."

notification/tests/test_tasks.py

@@ -7,6 +7,7 @@
 #
 
 import json
+import uuid
 from unittest.mock import patch
 
 from django.conf import settings
@@ -16,13 +17,54 @@
 from dje.models import Dataspace
 from dje.tests import create_superuser
 from notification.models import WebhookSubscription
+from notification.tasks import deliver_webhook_task
+from product_portfolio.tests import make_product
 from workflow.models import Priority
 from workflow.models import Question
 from workflow.models import Request
 from workflow.models import RequestComment
 from workflow.models import RequestTemplate
 
 
+class DeliverWebhookTaskTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.webhook = WebhookSubscription.objects.create(
+            dataspace=self.dataspace,
+            target_url="http://127.0.0.1:8000/",
+            event="policy.violation_detected",
+        )
+
+    @patch("notification.models.WebhookSubscription.deliver")
+    def test_deliver_webhook_task_resolves_secured_model_instance(self, mock_deliver):
+        product = make_product(self.dataspace)
+        deliver_webhook_task(
+            webhook_subscription_uuid=self.webhook.uuid,
+            payload_override={"text": "test"},
+            instance_app_label="product_portfolio",
+            instance_model_name="product",
+            instance_pk=product.pk,
+        )
+        mock_deliver.assert_called_once()
+        delivered_instance = mock_deliver.call_args[0][0]
+        self.assertEqual(product, delivered_instance)
+
+    def test_deliver_webhook_task_missing_subscription_logs_error(self):
+        with self.assertLogs("dje", level="ERROR") as captured:
+            deliver_webhook_task(webhook_subscription_uuid=uuid.uuid4())
+        self.assertTrue(any("not found" in line for line in captured.output))
+
+    def test_deliver_webhook_task_missing_instance_logs_error(self):
+        with self.assertLogs("dje", level="ERROR") as captured:
+            deliver_webhook_task(
+                webhook_subscription_uuid=self.webhook.uuid,
+                instance_app_label="product_portfolio",
+                instance_model_name="product",
+                instance_pk=99999999,
+            )
+        self.assertTrue(any("not found" in line for line in captured.output))
+
+
 class NotificationTasksTestCase(TestCase):
     def setUp(self):
         self.nexb_dataspace = Dataspace.objects.create(name="nexB")

policy/tasks.py

@@ -70,7 +70,7 @@ def evaluate_all_products_rules_task(include_locked=False, product_uuids=None):
     if product_uuids is not None:
         products = products.filter(uuid__in=product_uuids)
     elif not include_locked:
-        products = products.exclude_locked()
+        products = products.exclude(configuration_status__is_locked=True)
 
     count = products.count()
     logger.info(f"Starting policy rule evaluation for {count} product(s).")

policy/tests/test_tasks.py

@@ -0,0 +1,149 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# DejaCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: AGPL-3.0-only
+# See https://github.com/aboutcode-org/dejacode for support or download.
+# See https://aboutcode.org for more information about AboutCode FOSS projects.
+#
+
+import uuid
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from dje.models import Dataspace
+from policy.tasks import evaluate_all_products_rules_task
+from policy.tasks import evaluate_product_rules_task
+from policy.tasks import fire_policy_webhooks
+from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_status
+
+
+class FirePolicyWebhooksTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.tasks.fire_webhooks")
+    def test_fire_policy_webhooks_dispatches_violation_detected(self, mock_fire):
+        violation = MagicMock()
+        violation.rule_label = "Usage Policy Error"
+        violation.violation_count = 3
+        fire_policy_webhooks(self.product, new_violations=[violation], resolved_count=0)
+        mock_fire.assert_called_once()
+        event_name, kwargs = mock_fire.call_args[0][0], mock_fire.call_args[1]
+        self.assertEqual("policy.violation_detected", event_name)
+        self.assertIn("Policy violations detected", kwargs["payload_override"]["text"])
+        self.assertIn("Usage Policy Error", kwargs["payload_override"]["text"])
+
+    @patch("policy.tasks.fire_webhooks")
+    def test_fire_policy_webhooks_dispatches_violation_resolved(self, mock_fire):
+        fire_policy_webhooks(self.product, new_violations=[], resolved_count=2)
+        mock_fire.assert_called_once()
+        event_name, kwargs = mock_fire.call_args[0][0], mock_fire.call_args[1]
+        self.assertEqual("policy.violation_resolved", event_name)
+        self.assertIn("2 policy violation(s) resolved", kwargs["payload_override"]["text"])
+
+    @patch("policy.tasks.fire_webhooks")
+    def test_fire_policy_webhooks_dispatches_both_events(self, mock_fire):
+        violation = MagicMock()
+        violation.rule_label = "License Coverage Gap"
+        violation.violation_count = 1
+        fire_policy_webhooks(self.product, new_violations=[violation], resolved_count=1)
+        self.assertEqual(2, mock_fire.call_count)
+        events_fired = [c[0][0] for c in mock_fire.call_args_list]
+        self.assertIn("policy.violation_detected", events_fired)
+        self.assertIn("policy.violation_resolved", events_fired)
+
+    @patch("policy.tasks.fire_webhooks")
+    def test_fire_policy_webhooks_silent_when_no_changes(self, mock_fire):
+        fire_policy_webhooks(self.product, new_violations=[], resolved_count=0)
+        mock_fire.assert_not_called()
+
+
+class EvaluateProductRulesTaskTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_product_rules_task_runs_evaluation(self, mock_evaluate, mock_fire):
+        mock_evaluate.return_value = ([], 0)
+        evaluate_product_rules_task(product_uuid=self.product.uuid)
+        mock_evaluate.assert_called_once_with(self.product)
+        mock_fire.assert_called_once_with(self.product, [], 0)
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_product_rules_task_unknown_uuid_logs_error(self, mock_evaluate, mock_fire):
+        with self.assertLogs("policy.tasks", level="ERROR") as captured:
+            evaluate_product_rules_task(product_uuid=uuid.uuid4())
+        mock_evaluate.assert_not_called()
+        mock_fire.assert_not_called()
+        self.assertTrue(any("not found" in line for line in captured.output))
+
+
+class EvaluateAllProductsRulesTaskTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_all_products_excludes_locked_by_default(self, mock_evaluate, mock_fire):
+        # Regression: previously called .exclude_locked() on DataspacedQuerySet which lacks that
+        # method. Now uses .exclude(configuration_status__is_locked=True) inline.
+        mock_evaluate.return_value = ([], 0)
+        active_product = make_product(self.dataspace)
+        locked_status = make_product_status(self.dataspace, is_locked=True)
+        locked_product = make_product(self.dataspace, configuration_status=locked_status)
+        mock_evaluate.reset_mock()
+
+        evaluate_all_products_rules_task()
+
+        evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+        self.assertIn(active_product, evaluated_products)
+        self.assertNotIn(locked_product, evaluated_products)
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_all_products_includes_locked_when_requested(self, mock_evaluate, mock_fire):
+        mock_evaluate.return_value = ([], 0)
+        locked_status = make_product_status(self.dataspace, is_locked=True)
+        locked_product = make_product(self.dataspace, configuration_status=locked_status)
+        mock_evaluate.reset_mock()
+
+        evaluate_all_products_rules_task(include_locked=True)
+
+        evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+        self.assertIn(locked_product, evaluated_products)
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_all_products_filters_by_uuids(self, mock_evaluate, mock_fire):
+        mock_evaluate.return_value = ([], 0)
+        product_a = make_product(self.dataspace)
+        product_b = make_product(self.dataspace)
+        mock_evaluate.reset_mock()
+
+        evaluate_all_products_rules_task(product_uuids=[product_a.uuid])
+
+        evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+        self.assertIn(product_a, evaluated_products)
+        self.assertNotIn(product_b, evaluated_products)
+
+    @patch("policy.tasks.fire_policy_webhooks")
+    @patch("policy.tasks.evaluate_rules")
+    def test_evaluate_all_products_uuid_filter_ignores_locked_exclusion(
+        self, mock_evaluate, mock_fire
+    ):
+        mock_evaluate.return_value = ([], 0)
+        locked_status = make_product_status(self.dataspace, is_locked=True)
+        locked_product = make_product(self.dataspace, configuration_status=locked_status)
+        mock_evaluate.reset_mock()
+
+        evaluate_all_products_rules_task(product_uuids=[locked_product.uuid])
+
+        evaluated_products = [c[0][0] for c in mock_evaluate.call_args_list]
+        self.assertIn(locked_product, evaluated_products)