fix and add unit tests

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

Author: tdruez

product_portfolio/tests/test_admin.py

@@ -457,6 +457,39 @@ def test_product_admin_changeform_save_as_includes_relationships(self):
         self.assertEqual(1, new_product.productpackages.count())
         self.assertEqual(1, new_product.product_triage_rulesets.count())
 
+    @patch("product_portfolio.admin.reevaluate_products")
+    def test_product_admin_changeform_save_as_evaluates_the_new_product_once(self, mock_reevaluate):
+        # Cloning must evaluate the new product exactly once, not once per cloned
+        # ProductPackage relationship.
+        package2 = make_package(self.dataspace)
+        ProductPackage.objects.create(
+            product=self.product1, package=self.package1, dataspace=self.dataspace
+        )
+        ProductPackage.objects.create(
+            product=self.product1, package=package2, dataspace=self.dataspace
+        )
+        self.client.login(username=self.user.username, password="secret")
+
+        url = self.product1.get_admin_url()
+        data = {
+            "_saveasnew": "Save as new",
+            "name": self.product1.name,
+            "version": "new version 2",
+            "productcomponents-INITIAL_FORMS": 0,
+            "productcomponents-TOTAL_FORMS": 0,
+            "productpackages-INITIAL_FORMS": 0,
+            "productpackages-TOTAL_FORMS": 0,
+        }
+
+        response = self.client.post(url, data)
+
+        self.assertEqual(response.status_code, 302)
+        new_product = Product.unsecured_objects.get(
+            name=self.product1.name, version="new version 2"
+        )
+        self.assertEqual(2, new_product.productpackages.count())
+        mock_reevaluate.assert_called_once_with([new_product])
+
     def test_codebaseresource_admin_changeform_product_prefill_on_save_addanother(self):
         self.client.login(username=self.user.username, password="secret")
         url = reverse("admin:product_portfolio_codebaseresource_add")

product_portfolio/tests/test_importers.py

@@ -378,6 +378,28 @@ def test_productpackage_import_save_all(self):
         self.assertEqual(False, productpackage.is_deployed)
         self.assertEqual(True, productpackage.is_modified)
 
+    @mock.patch("product_portfolio.importers.reevaluate_product_triage_rulesets_task")
+    @mock.patch("product_portfolio.importers.evaluate_product_rules_task")
+    def test_productpackage_import_evaluates_each_product_once_for_multiple_rows(
+        self, mock_policy_task, mock_triage_task
+    ):
+        formset_data = self.package_formset_data.copy()
+        package1 = Package.objects.create(filename="p1.zip", dataspace=self.dataspace)
+        package2 = Package.objects.create(filename="p2.zip", dataspace=self.dataspace)
+
+        formset_data["form-TOTAL_FORMS"] = "2"
+        formset_data["form-0-package"] = package1.filename
+        formset_data["form-1-product"] = formset_data["form-0-product"]
+        formset_data["form-1-package"] = package2.filename
+
+        importer = ProductPackageImporter(self.super_user, formset_data=formset_data)
+        self.assertTrue(importer.formset.is_valid())
+        importer.save_all()
+
+        self.assertEqual(2, len(importer.results["added"]))
+        mock_policy_task.delay.assert_called_once_with(product_uuid=self.p1.uuid)
+        mock_triage_task.delay.assert_called_once_with(product_uuid=self.p1.uuid)
+
 
 class CodebaseResourceImporterTestCase(TestCase):
     def setUp(self):

product_portfolio/tests/test_views.py

@@ -4416,8 +4416,8 @@ def test_post_without_change_perm_returns_404(self):
         response = self.client.post(url)
         self.assertEqual(404, response.status_code)
 
-    @patch("vulnerabilities.triage.signals.evaluate_ruleset")
-    def test_post_assigns_the_submitted_rulesets(self, mock_evaluate):
+    @patch("product_portfolio.views.reevaluate_product_rulesets")
+    def test_post_assigns_the_submitted_rulesets(self, mock_reevaluate):
         self.client.login(username="nexb_user", password="secret")
         url = self.product1.get_manage_triage_rulesets_url()
         response = self.client.post(url, {"ruleset_uuids": [str(self.ruleset.uuid)]})
@@ -4427,14 +4427,14 @@ def test_post_assigns_the_submitted_rulesets(self, mock_evaluate):
                 product=self.product1, ruleset=self.ruleset
             ).exists()
         )
-        mock_evaluate.assert_called_once_with(ruleset=self.ruleset, product=self.product1)
+        mock_reevaluate.assert_called_once_with(self.product1)
 
-    @patch("vulnerabilities.triage.signals.evaluate_ruleset")
-    def test_post_unassigns_the_deselected_rulesets(self, mock_evaluate):
+    @patch("product_portfolio.views.delete_triage_records_for_assignment")
+    @patch("product_portfolio.views.reevaluate_product_rulesets")
+    def test_post_unassigns_the_deselected_rulesets(self, mock_reevaluate, mock_delete):
         ProductTriageRuleset.objects.create(
             product=self.product1, ruleset=self.ruleset, dataspace=self.dataspace
         )
-        mock_evaluate.reset_mock()
         self.client.login(username="nexb_user", password="secret")
         url = self.product1.get_manage_triage_rulesets_url()
         response = self.client.post(url, {"ruleset_uuids": []})
@@ -4444,7 +4444,8 @@ def test_post_unassigns_the_deselected_rulesets(self, mock_evaluate):
                 product=self.product1, ruleset=self.ruleset
             ).exists()
         )
-        mock_evaluate.assert_not_called()
+        mock_delete.assert_called_once_with(ruleset=self.ruleset, product=self.product1)
+        mock_reevaluate.assert_called_once_with(self.product1)
 
 
 class TabCompliancePolicyContextTestCase(TestCase):

vulnerabilities/triage/engine.py

@@ -96,7 +96,7 @@ def apply_preset_for_vulnerabilities(preset, product, vulnerability_ids):
         existing = existing_analyses.get((product_package_id, vulnerability_id))
 
         if existing is not None and existing.applied_by_preset_id is None:
-            continue  # Human-owned analysis -- never overwrite
+            continue  # User-owned analysis, never overwrite
 
         if existing is None:
             product_package = product_packages_by_id[product_package_id]

vulnerabilities/triage/tasks.py

@@ -29,8 +29,7 @@ def reevaluate_product_triage_rulesets_task(product_uuid):
         product = get_unsecured_manager(Product).get(uuid=product_uuid)
     except Product.DoesNotExist:
         logger.error(
-            f"reevaluate_product_triage_rulesets_task: product {product_uuid} not found,"
-            " skipping."
+            f"reevaluate_product_triage_rulesets_task: product {product_uuid} not found, skipping."
         )
         return
 

vulnerabilities/triage/tests/test_admin.py

@@ -0,0 +1,181 @@
+#
+# 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 django.contrib.contenttypes.models import ContentType
+from django.test import TestCase
+from django.urls import reverse
+
+from component_catalog.tests import make_package
+from dje.models import Dataspace
+from dje.tests import create_superuser
+from dje.tests import create_user
+from product_portfolio.models import Product
+from product_portfolio.tests import make_product
+from product_portfolio.tests import make_product_package
+from vulnerabilities.tests import make_vulnerability
+from vulnerabilities.triage.engine import evaluate_ruleset
+from vulnerabilities.triage.models import TriageAction
+from vulnerabilities.triage.models import TriageRecord
+from vulnerabilities.triage.tests import make_product_triage_ruleset
+from vulnerabilities.triage.tests import make_triage_ruleset
+from workflow.models import RequestTemplate
+
+RULE_CONFIG_DATA = {
+    "rule_risk_score_enabled": "on",
+    "rule_risk_score_min_risk_score": "8.0",
+}
+
+
+class TriageRulesetAdminSaveModelTestCase(TestCase):
+    def setUp(self):
+        self.dataspace = Dataspace.objects.create(name="nexB")
+        self.super_user = create_superuser("nexb_user", self.dataspace)
+        self.client.login(username="nexb_user", password="secret")
+        self.product = make_product(self.dataspace)
+        self.package = make_package(self.dataspace)
+        make_product_package(self.product, package=self.package)
+
+    def test_creating_a_ruleset_does_not_evaluate_any_product(self):
+        make_vulnerability(self.dataspace, affecting=self.package, risk_score=9.0)
+        url = reverse("admin:vulnerabilities_triage_triageruleset_add")
+        data = {
+            "name": "My Ruleset",
+            "precedence": 100,
+            "enabled": "on",
+            **RULE_CONFIG_DATA,
+        }
+
+        response = self.client.post(url, data)
+
+        self.assertEqual(302, response.status_code)
+        self.assertFalse(TriageRecord.objects.exists())
+
+    def test_updating_an_enabled_assigned_ruleset_reevaluates_its_products(self):
+        vulnerability = make_vulnerability(self.dataspace, affecting=self.package, risk_score=9.0)
+        ruleset = make_triage_ruleset(
+            self.dataspace,
+            recommended_action=TriageAction.UPGRADE,
+            enabled=True,
+        )
+        make_product_triage_ruleset(self.product, ruleset=ruleset)
+        self.assertFalse(TriageRecord.objects.exists())
+
+        url = ruleset.get_admin_url()
+        data = {
+            "name": ruleset.name,
+            "precedence": ruleset.precedence,
+            "recommended_action": TriageAction.UPGRADE,
+            "enabled": "on",
+            **RULE_CONFIG_DATA,
+        }
+
+        response = self.client.post(url, data)
+
+        self.assertEqual(302, response.status_code)
+        record = TriageRecord.objects.get()
+        self.assertEqual(vulnerability, record.vulnerability)
+
+    def test_disabling_a_ruleset_deletes_its_triage_records(self):
+        make_vulnerability(self.dataspace, affecting=self.package, risk_score=9.0)
+        ruleset = make_triage_ruleset(
+            self.dataspace,
+            recommended_action=TriageAction.UPGRADE,
+            enabled=True,
+            rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}},
+        )
+        make_product_triage_ruleset(self.product, ruleset=ruleset)
+        evaluate_ruleset(ruleset, self.product)
+        self.assertTrue(TriageRecord.objects.exists())
+
+        url = ruleset.get_admin_url()
+        data = {
+            "name": ruleset.name,
+            "precedence": ruleset.precedence,
+            "recommended_action": TriageAction.UPGRADE,
+            # "enabled" omitted: disables the ruleset.
+            **RULE_CONFIG_DATA,
+        }
+
+        response = self.client.post(url, data)
+
+        self.assertEqual(302, response.status_code)
+        self.assertFalse(TriageRecord.objects.exists())
+
+    def test_disabling_a_ruleset_keeps_records_that_have_an_open_request(self):
+        vulnerability = make_vulnerability(self.dataspace, affecting=self.package, risk_score=9.0)
+        requester = create_user("requester", self.dataspace)
+        request_template = RequestTemplate.objects.create(
+            name="Template",
+            description="Header",
+            dataspace=self.dataspace,
+            content_type=ContentType.objects.get_for_model(Product),
+            created_by=requester,
+        )
+        ruleset = make_triage_ruleset(
+            self.dataspace,
+            recommended_action=TriageAction.UPGRADE,
+            enabled=True,
+            request_template=request_template,
+            rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}},
+        )
+        make_product_triage_ruleset(self.product, ruleset=ruleset)
+        evaluate_ruleset(ruleset, self.product)
+        record = TriageRecord.objects.get()
+        self.assertIsNotNone(record.request)
+
+        url = ruleset.get_admin_url()
+        data = {
+            "name": ruleset.name,
+            "precedence": ruleset.precedence,
+            "recommended_action": TriageAction.UPGRADE,
+            # "enabled" omitted: disables the ruleset.
+            **RULE_CONFIG_DATA,
+        }
+
+        response = self.client.post(url, data)
+
+        self.assertEqual(302, response.status_code)
+        record.refresh_from_db()
+        self.assertEqual(vulnerability, record.vulnerability)
+
+    def test_disabling_then_reenabling_a_ruleset_reuses_the_existing_request(self):
+        # Regression: disabling then re-enabling a ruleset used to reopen a new Request
+        # instead of reconnecting to the one already tracking this vulnerability.
+        make_vulnerability(self.dataspace, affecting=self.package, risk_score=9.0)
+        requester = create_user("requester", self.dataspace)
+        request_template = RequestTemplate.objects.create(
+            name="Template",
+            description="Header",
+            dataspace=self.dataspace,
+            content_type=ContentType.objects.get_for_model(Product),
+            created_by=requester,
+        )
+        ruleset = make_triage_ruleset(
+            self.dataspace,
+            recommended_action=TriageAction.UPGRADE,
+            enabled=True,
+            request_template=request_template,
+            rules_config={"risk_score": {"is_active": True, "min_risk_score": 8.0}},
+        )
+        make_product_triage_ruleset(self.product, ruleset=ruleset)
+        evaluate_ruleset(ruleset, self.product)
+        original_request = TriageRecord.objects.get().request
+        self.assertIsNotNone(original_request)
+
+        url = ruleset.get_admin_url()
+        base_data = {
+            "name": ruleset.name,
+            "precedence": ruleset.precedence,
+            "recommended_action": TriageAction.UPGRADE,
+            **RULE_CONFIG_DATA,
+        }
+        self.client.post(url, base_data)  # Disable.
+        self.client.post(url, {**base_data, "enabled": "on"})  # Re-enable.
+
+        self.assertEqual(1, ruleset.triage_records.count())
+        self.assertEqual(original_request, TriageRecord.objects.get().request)