Fix #603 unify product cloning

product_portfolio/admin.py

@@ -7,7 +7,6 @@
 #
 
 import copy
-import uuid
 
 from django import forms
 from django.contrib import admin
@@ -59,8 +58,6 @@
 from product_portfolio.importers import CodebaseResourceImporter
 from product_portfolio.importers import ProductComponentImporter
 from product_portfolio.importers import ProductPackageImporter
-from product_portfolio.importers import paused_product_package_reevaluation
-from product_portfolio.importers import reevaluate_products
 from product_portfolio.inlines import CodebaseResourceUsageDeployedFromInline
 from product_portfolio.inlines import CodebaseResourceUsageDeployedToInline
 from product_portfolio.inlines import ProductComponentInline
@@ -74,7 +71,6 @@
 from product_portfolio.models import ProductRelationStatus
 from product_portfolio.models import ProductStatus
 from reporting.filters import ReportingQueryListFilter
-from vulnerabilities.triage.models import ProductTriageRuleset
 
 
 @admin.register(ProductStatus, site=dejacode_site)
@@ -230,28 +226,11 @@ def has_delete_permission(self, request, obj=None):
         return super().has_delete_permission(request, obj)
 
     def save_model(self, request, obj, form, change):
-        """
-        Add view/change/delete Object permissions to the Product creator.
-
-        Add the copy of the ProductComponent and ProductPackage during a `save_as` since
-        those value are not in the form as inlines.
-        """
+        """Add view/change/delete Object permissions to the Product creator."""
         super().save_model(request, obj, form, change)
         if not change:
             assign_all_object_permissions(request.user, obj)
 
-        if "_saveasnew" in request.POST:
-            old_product_id = request.resolver_match.kwargs.get("object_id")
-            old_product = self.get_object(request, old_product_id)
-            with paused_product_package_reevaluation():
-                for model_class in [ProductComponent, ProductPackage, ProductTriageRuleset]:
-                    for relationship in model_class.objects.filter(product=old_product):
-                        relationship.id = None
-                        relationship.uuid = uuid.uuid4()
-                        relationship.product = obj
-                        relationship.save()
-            reevaluate_products([obj])
-
     def get_obj_perms_user_select_form(self, request):
         """
         Return a UserManageForm class using the request object for User Dataspace scoping.
@@ -390,6 +369,7 @@ class ProductAdmin(
         ProductPackageInline,
     ]
     form = ProductAdminForm
+    save_as = False
     actions = ["evaluate_policy_rules"]
     actions_to_remove = ["copy_to", "compare_with", "delete_selected"]
     navigation_buttons = True

product_portfolio/api.py

@@ -562,8 +562,9 @@ def clone(self, request, *args, **kwargs):
         Clone this Product into a new one.
 
         Always copies the base Product fields under the given name/version.
-        Optionally copies the inventory (Components, Packages, and Dependencies),
-        Codebase resources, Vulnerability triage rules, and object Permissions,
+        Optionally copies the inventory (Components, Packages, Dependencies, and
+        vulnerability analyses), Codebase resources and deployment links, Vulnerability
+        triage rules, and object Permissions,
         depending on the submitted flags (all default to true).
         """
         product = self.get_object()

product_portfolio/forms.py

@@ -13,6 +13,7 @@
 from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.db import transaction
+from django.db.models import Q
 from django.forms import BaseModelFormSet
 from django.forms.formsets import DELETION_FIELD_NAME
 from django.urls import reverse_lazy
@@ -56,6 +57,7 @@
 from dje.widgets import AwesompleteInputWidget
 from dje.widgets import DatePicker
 from product_portfolio.models import CodebaseResource
+from product_portfolio.models import CodebaseResourceUsage
 from product_portfolio.models import Product
 from product_portfolio.models import ProductComponent
 from product_portfolio.models import ProductComponentAssignedLicense
@@ -65,6 +67,7 @@
 from product_portfolio.models import ScanCodeProject
 from product_portfolio.tasks import pull_project_data_from_scancodeio_task
 from product_portfolio.tasks import scancodeio_submit_project_task
+from vulnerabilities.models import VulnerabilityAnalysis
 from vulnerabilities.triage.models import ProductTriageRuleset
 
 
@@ -133,13 +136,7 @@ class ProductForm(
     DataspacedModelForm,
 ):
     default_on_addition_fields = ["configuration_status"]
-    save_as = True
-    clone_m2m_classes = [
-        ProductComponent,
-        ProductPackage,
-        CodebaseResource,
-        ProductTriageRuleset,
-    ]
+    save_as = False
 
     keywords = KeywordsField()
 
@@ -196,11 +193,6 @@ def assign_object_perms(self, user):
         if user.has_perm("product_portfolio.delete_product"):
             assign_perm("delete_product", user, self.instance)
 
-    def _clone_m2m(self, original_instance_id, cloned_instance):
-        cloned_pairs_by_model = super()._clone_m2m(original_instance_id, cloned_instance)
-        relink_cloned_codebase_resources(cloned_pairs_by_model)
-        return cloned_pairs_by_model
-
     @property
     def helper(self):
         helper = super().helper
@@ -221,7 +213,6 @@ def helper(self):
                 Group("vulnerabilities_risk_threshold", HTML(""), HTML("")),
                 HTML("<hr>"),
                 Submit("submit", self.submit_label, css_class="btn-success"),
-                self.save_as_new_submit,
             ),
         )
 
@@ -341,6 +332,52 @@ def _clone_relations(self, model_class, target_product):
 
         return cloned_pairs
 
+    def _clone_vulnerability_analyses(self, cloned_pairs_by_model, target_product):
+        cloned_package_by_original_pk = dict(cloned_pairs_by_model.get(ProductPackage, []))
+        if not cloned_package_by_original_pk:
+            return
+
+        for analysis in VulnerabilityAnalysis.objects.filter(product=self.source_product):
+            cloned_product_package = cloned_package_by_original_pk.get(analysis.product_package_id)
+            if not cloned_product_package:
+                continue
+
+            VulnerabilityAnalysis.objects.create(
+                dataspace=target_product.dataspace,
+                product_package=cloned_product_package,
+                vulnerability=analysis.vulnerability,
+                state=analysis.state,
+                justification=analysis.justification,
+                responses=analysis.responses,
+                is_reachable=analysis.is_reachable,
+                detail=analysis.detail,
+                created_by=self.user,
+                last_modified_by=self.user,
+            )
+
+    def _clone_deployment_links(self, cloned_pairs_by_model, target_product):
+        cloned_resource_by_original_pk = dict(cloned_pairs_by_model.get(CodebaseResource, []))
+        if not cloned_resource_by_original_pk:
+            return
+
+        original_resource_ids = set(cloned_resource_by_original_pk)
+        usages = CodebaseResourceUsage.objects.filter(
+            Q(deployed_from_id__in=original_resource_ids)
+            | Q(deployed_to_id__in=original_resource_ids)
+        )
+        for usage in usages:
+            deployed_from = cloned_resource_by_original_pk.get(usage.deployed_from_id)
+            deployed_to = cloned_resource_by_original_pk.get(usage.deployed_to_id)
+            if not deployed_from or not deployed_to:
+                continue
+
+            CodebaseResourceUsage.objects.create(
+                uuid=uuid.uuid4(),
+                dataspace=target_product.dataspace,
+                deployed_from=deployed_from,
+                deployed_to=deployed_to,
+            )
+
     def save(self, commit=True):
         from product_portfolio.importers import paused_product_package_reevaluation
         from product_portfolio.importers import reevaluate_products
@@ -359,8 +396,14 @@ def save(self, commit=True):
                     cloned_pairs = self._clone_relations(model_class, instance)
                     cloned_pairs_by_model[model_class] = cloned_pairs
 
+            if self.cleaned_data.get("copy_inventory"):
+                self._clone_vulnerability_analyses(cloned_pairs_by_model, instance)
+
         relink_cloned_codebase_resources(cloned_pairs_by_model)
 
+        if self.cleaned_data.get("copy_codebase_resources"):
+            self._clone_deployment_links(cloned_pairs_by_model, instance)
+
         reevaluate_products([instance])
 
         if self.cleaned_data.get("copy_object_permissions"):

product_portfolio/tests/test_admin.py

@@ -25,8 +25,6 @@
 from product_portfolio.models import ProductComponent
 from product_portfolio.models import ProductPackage
 from product_portfolio.tests import make_product_dependency
-from vulnerabilities.triage.models import ProductTriageRuleset
-from vulnerabilities.triage.models import TriageRuleset
 
 
 class ProductPortfolioAdminsTestCase(TestCase):
@@ -178,16 +176,6 @@ def test_product_admin_changeform_unique_validation(self):
         expected = {NON_FIELD_ERRORS: [error_msg]}
         self.assertEqual(expected, response.context_data["adminform"].form.errors)
 
-        url = self.product1.get_admin_url()
-        data["_saveasnew"] = "Save as new"
-        response = self.client.post(url, data)
-        self.assertContains(response, error_msg)
-        self.assertEqual(expected, response.context_data["adminform"].form.errors)
-
-        data["version"] = "1.0"
-        response = self.client.post(url, data)
-        self.assertEqual(response.status_code, 302)
-
     def test_productcomponent_admin_changeform_feature_datalist(self):
         self.client.login(username=self.user.username, password="secret")
         url = reverse("admin:product_portfolio_productcomponent_add")
@@ -423,72 +411,13 @@ def test_productcomponent_admin_check_for_update_actions(self):
         self.assertIn(expected1, actions)
         self.assertIn(expected2, actions)
 
-    def test_product_admin_changeform_save_as_includes_relationships(self):
-        self.client.login(username=self.user.username, password="secret")
-
-        ProductComponent.objects.create(
-            product=self.product1, component=self.component1, dataspace=self.dataspace
-        )
-        ProductPackage.objects.create(
-            product=self.product1, package=self.package1, dataspace=self.dataspace
-        )
-        ruleset = TriageRuleset.objects.create(
-            name="Upgrade Ruleset", precedence=100, dataspace=self.dataspace
-        )
-        ProductTriageRuleset.objects.create(
-            product=self.product1, ruleset=ruleset, dataspace=self.dataspace
-        )
+    def test_product_admin_changeform_does_not_expose_save_as_new(self):
+        self.client.force_login(self.user)
 
         url = self.product1.get_admin_url()
-        data = {
-            "_saveasnew": "Save as new",
-            "name": self.product1.name,
-            "version": "new version",
-            "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")
-        self.assertEqual(1, new_product.productcomponents.count())
-        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])
+        response = self.client.get(url)
+        self.assertNotContains(response, 'name="_saveasnew"')
+        self.assertNotContains(response, "Save as new")
 
     def test_codebaseresource_admin_changeform_product_prefill_on_save_addanother(self):
         self.client.login(username=self.user.username, password="secret")

product_portfolio/tests/test_api.py

@@ -583,8 +583,26 @@ def test_api_product_endpoint_import_manifests_action(self):
 
     def test_api_product_endpoint_clone_action(self):
         url = reverse("api_v2:product-clone", args=[self.product1.uuid])
-        make_product_package(self.product1)
+        product_package = make_product_package(self.product1)
         make_product_triage_ruleset(self.product1)
+        vulnerability = make_vulnerability(
+            self.product1.dataspace, affecting=[product_package.package]
+        )
+        source_analysis = make_vulnerability_analysis(product_package, vulnerability)
+        deployed_from = CodebaseResource.objects.create(
+            product=self.product1, path="/src/", dataspace=self.product1.dataspace
+        )
+        deployed_to = CodebaseResource.objects.create(
+            product=self.product1,
+            path="/deploy/",
+            is_deployment_path=True,
+            dataspace=self.product1.dataspace,
+        )
+        source_usage = CodebaseResourceUsage.objects.create(
+            dataspace=self.product1.dataspace,
+            deployed_from=deployed_from,
+            deployed_to=deployed_to,
+        )
 
         self.client.login(username=self.base_user.username, password="secret")
         response = self.client.get(url)
@@ -615,6 +633,15 @@ def test_api_product_endpoint_clone_action(self):
         self.assertEqual(str(cloned_product.uuid), response.data["uuid"])
         self.assertEqual(1, cloned_product.productpackages.count())
         self.assertEqual(1, cloned_product.product_triage_rulesets.count())
+        cloned_analysis = cloned_product.vulnerability_analyses.get()
+        self.assertEqual(cloned_product.productpackages.get(), cloned_analysis.product_package)
+        self.assertEqual(source_analysis.state, cloned_analysis.state)
+        self.assertEqual(source_analysis.justification, cloned_analysis.justification)
+        self.assertEqual(source_analysis.responses, cloned_analysis.responses)
+        self.assertEqual(source_analysis.detail, cloned_analysis.detail)
+        cloned_usage = CodebaseResourceUsage.objects.get(deployed_from__product=cloned_product)
+        self.assertNotEqual(source_usage.deployed_from_id, cloned_usage.deployed_from_id)
+        self.assertNotEqual(source_usage.deployed_to_id, cloned_usage.deployed_to_id)
 
         # Submitting the same name/version again is rejected
         response = self.client.post(url, data)
@@ -625,13 +652,19 @@ def test_api_product_endpoint_clone_action(self):
             "name": self.product1.name,
             "version": "3.0",
             "copy_inventory": False,
+            "copy_codebase_resources": False,
             "copy_triage_rulesets": False,
         }
         response = self.client.post(url, data)
         self.assertEqual(status.HTTP_201_CREATED, response.status_code)
         empty_clone = Product.objects.get_queryset(self.base_user).get(version="3.0")
         self.assertEqual(0, empty_clone.productpackages.count())
         self.assertEqual(0, empty_clone.product_triage_rulesets.count())
+        self.assertEqual(0, empty_clone.vulnerability_analyses.count())
+        self.assertEqual(0, empty_clone.codebaseresources.count())
+        self.assertEqual(
+            0, CodebaseResourceUsage.objects.filter(deployed_from__product=empty_clone).count()
+        )
 
     def test_api_product_endpoint_import_from_scan_action(self):
         url = reverse("api_v2:product-import-from-scan", args=[self.product1.uuid])