add unit tests

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

Author: tdruez

dje/tests/test_admin.py

@@ -13,10 +13,12 @@
 from django.test import override_settings
 from django.urls import reverse
 
+from dje.admin import PolicyRulesConfigurationForm
 from dje.copier import copy_object
 from dje.filters import DataspaceFilter
 from dje.filters import MissingInFilter
 from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
 from dje.models import History
 from dje.search import advanced_search
 from dje.tests import add_perm
@@ -25,6 +27,7 @@
 from dje.tests import create_superuser
 from dje.tests import create_user
 from organization.models import Owner
+from policy.rules import RULE_REGISTRY
 
 
 class DataspacedModelAdminTestCase(TestCase):
@@ -613,3 +616,78 @@ def test_admin_group_permission_export_csv(self):
             'attachment; filename="dejacode_group_permission.csv"', response["Content-Disposition"]
         )
         self.assertEqual(b",change_license\r\nchange license,X\r\n", response.content)
+
+
+class PolicyRulesConfigurationFormTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.config = DataspaceConfiguration.objects.create(dataspace=self.dataspace)
+        self.Form = PolicyRulesConfigurationForm
+
+    def _bound_form(self, extra_data=None):
+        data = {}
+        for rule_type in RULE_REGISTRY:
+            data[f"rule_{rule_type}_enabled"] = False
+            data[f"rule_{rule_type}_threshold"] = ""
+        if extra_data:
+            data.update(extra_data)
+        return self.Form(data=data, instance=self.config)
+
+    def test_enabled_rule_included_in_config(self):
+        form = self._bound_form({"rule_usage_policy_error_enabled": True})
+        self.assertTrue(form.is_valid(), form.errors)
+        result = form.build_policy_rules_config()
+        self.assertIn("usage_policy_error", result)
+        self.assertTrue(result["usage_policy_error"]["is_active"])
+
+    def test_disabled_rule_not_in_config(self):
+        form = self._bound_form()
+        self.assertTrue(form.is_valid(), form.errors)
+        result = form.build_policy_rules_config()
+        self.assertEqual({}, result)
+
+    def test_threshold_saved_when_set(self):
+        form = self._bound_form(
+            {
+                "rule_usage_policy_error_enabled": True,
+                "rule_usage_policy_error_threshold": "3",
+            }
+        )
+        self.assertTrue(form.is_valid(), form.errors)
+        result = form.build_policy_rules_config()
+        self.assertEqual(3, result["usage_policy_error"]["threshold"])
+
+    def test_initial_values_loaded_from_existing_config(self):
+        self.config.policy_rules_config = {
+            "usage_policy_error": {"is_active": True, "threshold": 7}
+        }
+        self.config.save()
+        form = self.Form(instance=self.config)
+        self.assertTrue(form.fields["rule_usage_policy_error_enabled"].initial)
+        self.assertEqual(7, form.fields["rule_usage_policy_error_threshold"].initial)
+        self.assertFalse(form.fields["rule_license_coverage_gap_enabled"].initial)
+
+    def test_save_persists_policy_rules_config_to_db(self):
+        form = self._bound_form({"rule_usage_policy_error_enabled": True})
+        self.assertTrue(form.is_valid(), form.errors)
+        form.save()
+        self.config.refresh_from_db()
+        self.assertTrue(self.config.policy_rules_config["usage_policy_error"]["is_active"])
+
+    def test_inline_shows_rule_fieldsets_on_existing_dataspace(self):
+        self.super_user = create_superuser("super_user", self.dataspace)
+        self.client.login(username="super_user", password="secret")
+        url = reverse("admin:dje_dataspace_change", args=[self.dataspace.pk])
+        response = self.client.get(url)
+        self.assertEqual(200, response.status_code)
+        self.assertContains(response, "Policy Rules Configuration")
+        for rule_type in RULE_REGISTRY:
+            self.assertContains(response, f"rule_{rule_type}_enabled")
+
+    def test_inline_not_shown_on_dataspace_add(self):
+        self.super_user = create_superuser("super_user", self.dataspace)
+        self.client.login(username="super_user", password="secret")
+        url = reverse("admin:dje_dataspace_add")
+        response = self.client.get(url)
+        self.assertEqual(200, response.status_code)
+        self.assertNotContains(response, "rule_usage_policy_error_enabled")

policy/tests/test_engine.py

@@ -11,7 +11,10 @@
 from django.test import TestCase
 
 from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
 from policy.engine import evaluate_rule
+from policy.engine import evaluate_rules
+from policy.engine import get_effective_config
 from product_portfolio.models import ProductPolicyViolation
 from product_portfolio.tests import make_product
 
@@ -71,3 +74,69 @@ def test_evaluate_rule_updates_count_on_existing_active_violation(self):
         self.assertFalse(created)
         self.assertEqual(7, violation.violation_count)
         self.assertEqual(1, ProductPolicyViolation.objects.count())
+
+
+class GetEffectiveConfigTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.product = make_product(self.dataspace)
+
+    def test_returns_defaults_when_no_dataspace_configuration(self):
+        config = get_effective_config("usage_policy_error", self.dataspace)
+        self.assertFalse(config["is_active"])
+        self.assertEqual(0, config["threshold"])
+        self.assertEqual({}, config["parameters"])
+
+    def test_reads_dataspace_override(self):
+        DataspaceConfiguration.objects.create(
+            dataspace=self.dataspace,
+            policy_rules_config={"usage_policy_error": {"is_active": True, "threshold": 5}},
+        )
+        self.dataspace.refresh_from_db()
+        config = get_effective_config("usage_policy_error", self.dataspace)
+        self.assertTrue(config["is_active"])
+        self.assertEqual(5, config["threshold"])
+
+    def test_falls_back_to_code_defaults_for_partial_config(self):
+        DataspaceConfiguration.objects.create(
+            dataspace=self.dataspace,
+            policy_rules_config={"usage_policy_error": {"is_active": True}},
+        )
+        self.dataspace.refresh_from_db()
+        config = get_effective_config("usage_policy_error", self.dataspace)
+        self.assertTrue(config["is_active"])
+        self.assertEqual(0, config["threshold"])
+
+
+class EvaluateRulesTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        DataspaceConfiguration.objects.create(
+            dataspace=self.dataspace,
+            policy_rules_config={"usage_policy_error": {"is_active": True}},
+        )
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.engine.fire_policy_webhooks")
+    @patch("policy.rules.UsagePolicyErrorRule.count_violations", return_value=3)
+    def test_evaluate_rules_creates_violation_for_active_rule(self, _mock_count, mock_fire):
+        new_violations, resolved_count = evaluate_rules(self.product)
+        self.assertEqual(1, len(new_violations))
+        self.assertEqual(0, resolved_count)
+        self.assertEqual(1, ProductPolicyViolation.objects.filter(resolved=False).count())
+        mock_fire.assert_called_once()
+
+    @patch("policy.engine.fire_policy_webhooks")
+    def test_evaluate_rules_resolves_violation_when_rule_inactive(self, _mock_fire):
+        ProductPolicyViolation.objects.create(
+            product=self.product,
+            dataspace=self.dataspace,
+            rule_type="license_coverage_gap",
+            violation_count=2,
+        )
+        new_violations, resolved_count = evaluate_rules(self.product)
+        self.assertEqual(0, len(new_violations))
+        self.assertGreater(resolved_count, 0)
+        self.assertTrue(
+            ProductPolicyViolation.objects.get(rule_type="license_coverage_gap").resolved
+        )

policy/tests/test_signals.py

@@ -0,0 +1,113 @@
+#
+# 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.
+#
+
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from component_catalog.tests import make_package
+from dje.models import Dataspace
+from dje.models import DataspaceConfiguration
+from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_package
+
+
+class ProductSaveSignalTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+
+    @patch("policy.signals.evaluate_product_rules_task.delay")
+    def test_product_save_queues_task(self, mock_delay):
+        product = make_product(self.dataspace)
+        mock_delay.assert_called_with(product_uuid=product.uuid)
+
+    @patch("policy.signals.evaluate_product_rules_task.delay")
+    def test_product_update_queues_task(self, mock_delay):
+        product = make_product(self.dataspace)
+        mock_delay.reset_mock()
+        product.save()
+        mock_delay.assert_called_once_with(product_uuid=product.uuid)
+
+
+class ProductPackageSignalTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.signals.evaluate_product_rules_task.delay")
+    def test_productpackage_save_queues_task(self, mock_delay):
+        make_product_package(self.product)
+        mock_delay.assert_called_with(product_uuid=self.product.uuid)
+
+    @patch("policy.signals.evaluate_product_rules_task.delay")
+    def test_productpackage_delete_queues_task(self, mock_delay):
+        pp = make_product_package(self.product)
+        mock_delay.reset_mock()
+        pp.delete()
+        mock_delay.assert_called_once_with(product_uuid=self.product.uuid)
+
+
+class PackageSaveSignalTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_package_create_does_not_queue_task(self, mock_delay):
+        make_package(self.dataspace)
+        mock_delay.assert_not_called()
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_package_update_with_products_queues_task(self, mock_delay):
+        package = make_package(self.dataspace)
+        make_product_package(self.product, package=package)
+        mock_delay.reset_mock()
+        package.save()
+        mock_delay.assert_called_once()
+        called_uuids = mock_delay.call_args[1]["product_uuids"]
+        self.assertIn(self.product.uuid, called_uuids)
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_package_update_without_products_does_not_queue_task(self, mock_delay):
+        package = make_package(self.dataspace)
+        mock_delay.reset_mock()
+        package.save()
+        mock_delay.assert_not_called()
+
+
+class DataspaceConfigurationSignalTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.config = DataspaceConfiguration.objects.create(dataspace=self.dataspace)
+        self.product = make_product(self.dataspace)
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_full_save_queues_task(self, mock_delay):
+        self.config.save()
+        mock_delay.assert_called_once()
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_policy_rules_config_update_fields_queues_task(self, mock_delay):
+        self.config.policy_rules_config = {"usage_policy_error": {"is_active": True}}
+        self.config.save(update_fields=["policy_rules_config"])
+        mock_delay.assert_called_once()
+        called_uuids = mock_delay.call_args[1]["product_uuids"]
+        self.assertIn(self.product.uuid, called_uuids)
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_unrelated_update_fields_does_not_queue_task(self, mock_delay):
+        self.config.save(update_fields=["scancodeio_url"])
+        mock_delay.assert_not_called()
+
+    @patch("policy.signals.evaluate_all_products_rules_task.delay")
+    def test_no_products_in_dataspace_does_not_queue_task(self, mock_delay):
+        empty_dataspace = Dataspace.objects.create(name="Empty")
+        empty_config = DataspaceConfiguration.objects.create(dataspace=empty_dataspace)
+        mock_delay.reset_mock()
+        empty_config.save(update_fields=["policy_rules_config"])
+        mock_delay.assert_not_called()

product_portfolio/models.py

@@ -432,7 +432,7 @@ class Product(
     # WARNING: Bypass the security system implemented in ProductSecuredManager.
     # This is to be used only in a few cases where the User scoping is not appropriated.
     # For example: `self.dataspace.product_set(manager='unsecured_objects').count()`
-    unsecured_objects = DataspacedManager()
+    unsecured_objects = DataspacedManager.from_queryset(ProductQuerySet)()
 
     class Meta(BaseProductMixin.Meta):
         permissions = (("view_product", "Can view product"),)

product_portfolio/tests/test_admin.py

@@ -6,6 +6,8 @@
 # See https://aboutcode.org for more information about AboutCode FOSS projects.
 #
 
+from unittest.mock import patch
+
 from django.core.exceptions import NON_FIELD_ERRORS
 from django.test import TestCase
 from django.urls import NoReverseMatch
@@ -101,7 +103,11 @@ def test_product_admin_form_clean_license_expression_in_alternate_dataspace(self
     def test_product_security_admin_changelist_available_actions(self):
         self.client.login(username=self.user.username, password="secret")
         response = self.client.get(self.product_changelist_url)
-        expected = [("", "---------"), ("mass_update", "Mass update")]
+        expected = [
+            ("", "---------"),
+            ("evaluate_policy_rules", "Evaluate policy rules"),
+            ("mass_update", "Mass update"),
+        ]
         self.assertEqual(expected, response.context_data["action_form"].fields["action"].choices)
 
         with self.assertRaises(NoReverseMatch):
@@ -539,3 +545,26 @@ def test_product_dependencies_add_view(self):
         dependency = self.product1.dependencies.get()
         self.assertEqual(self.package1, dependency.for_package)
         self.assertEqual(package2, dependency.resolved_to_package)
+
+
+class EvaluatePolicyRulesActionTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.super_user = create_superuser("nexb_user", self.dataspace)
+        self.product1 = Product.objects.create(name="Product1", dataspace=self.dataspace)
+        self.product2 = Product.objects.create(name="Product2", dataspace=self.dataspace)
+
+    @patch("product_portfolio.admin.evaluate_product_rules_task.delay")
+    def test_evaluate_policy_rules_action_queues_task_for_selected_products(self, mock_delay):
+        self.client.login(username="nexb_user", password="secret")
+        url = reverse("admin:product_portfolio_product_changelist")
+        data = {
+            "action": "evaluate_policy_rules",
+            "_selected_action": [self.product1.pk, self.product2.pk],
+        }
+        response = self.client.post(url, data, follow=True)
+        self.assertEqual(200, response.status_code)
+        self.assertEqual(2, mock_delay.call_count)
+        called_uuids = {call[1]["product_uuid"] for call in mock_delay.call_args_list}
+        self.assertIn(self.product1.uuid, called_uuids)
+        self.assertIn(self.product2.uuid, called_uuids)

product_portfolio/tests/test_admin_guardian.py

@@ -58,7 +58,11 @@ def test_product_guardian_admin_security_attributes(self):
 
         # actions = []
         # actions_to_remove = ['copy_to', 'compare_with', 'delete_selected']
-        expected = [("", "---------"), ("mass_update", "Mass update")]
+        expected = [
+            ("", "---------"),
+            ("evaluate_policy_rules", "Evaluate policy rules"),
+            ("mass_update", "Mass update"),
+        ]
         self.assertEqual(expected, response.context_data["action_form"].fields["action"].choices)
 
         # activity_log = False