fix: relink cloned codebase resources to the cloned inventory #601

dje/forms.py

@@ -209,14 +209,21 @@ def _clone_m2m(self, original_instance_id, cloned_instance):
         """
         Clone each defined m2m relations in `self.clone_m2m_classes` during
         the `save_as_new` process.
+
+        Return the (original_pk, cloned_instance) pairs grouped by model class.
         """
         field_name = self.instance._meta.model_name
 
+        cloned_pairs_by_model = {}
         for model_class in self.clone_m2m_classes:
             if model_class.__name__ == "Subcomponent":
                 field_name = "parent"
 
-            clone_related_objects(model_class, field_name, original_instance_id, cloned_instance)
+            cloned_pairs_by_model[model_class] = clone_related_objects(
+                model_class, field_name, original_instance_id, cloned_instance
+            )
+
+        return cloned_pairs_by_model
 
     @property
     def save_as_new_submit(self):

product_portfolio/forms.py

@@ -102,6 +102,30 @@ def clean(self):
             raise self.instance.unique_error_message(model, unique_check)
 
 
+def relink_cloned_codebase_resources(cloned_pairs_by_model):
+    """
+    Point the cloned codebase resources to the cloned product component and package.
+    A link to a relation that was not cloned is cleared, as it belongs to the source product.
+    """
+    cloned_component_by_original_pk = dict(cloned_pairs_by_model.get(ProductComponent, []))
+    cloned_package_by_original_pk = dict(cloned_pairs_by_model.get(ProductPackage, []))
+
+    linked_resources = [
+        cloned_resource
+        for _, cloned_resource in cloned_pairs_by_model.get(CodebaseResource, [])
+        if cloned_resource.product_component_id or cloned_resource.product_package_id
+    ]
+    for resource in linked_resources:
+        resource.product_component = cloned_component_by_original_pk.get(
+            resource.product_component_id
+        )
+        resource.product_package = cloned_package_by_original_pk.get(resource.product_package_id)
+
+    CodebaseResource.objects.bulk_update(
+        linked_resources, ["product_component", "product_package"], batch_size=1000
+    )
+
+
 class ProductForm(
     LicenseExpressionFormMixin,
     DefaultOnAdditionLabelMixin,
@@ -172,6 +196,11 @@ 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
@@ -291,7 +320,11 @@ def helper(self):
         return helper
 
     def _clone_relations(self, model_class, target_product):
-        """Clone `model_class` relations from the source product onto `target_product`."""
+        """
+        Clone `model_class` relations from the source product onto `target_product`.
+
+        Return a list of (original_pk, cloned_relation) pairs.
+        """
         assigned_license_model = self.assigned_license_models.get(model_class)
         save_kwargs = {"copy": True} if assigned_license_model else None
 
@@ -300,12 +333,14 @@ def _clone_relations(self, model_class, target_product):
         )
 
         if not assigned_license_model:
-            return
+            return cloned_pairs
 
         license_model, license_fk_name = assigned_license_model
         for original_pk, cloned_relation in cloned_pairs:
             clone_related_objects(license_model, license_fk_name, original_pk, cloned_relation)
 
+        return cloned_pairs
+
     def save(self, commit=True):
         from product_portfolio.importers import paused_product_package_reevaluation
         from product_portfolio.importers import reevaluate_products
@@ -315,12 +350,16 @@ def save(self, commit=True):
         History.log_addition(self.user, instance)
         assign_all_object_permissions(self.user, instance)
 
+        cloned_pairs_by_model = {}
         with paused_product_package_reevaluation():
             for field_name, model_classes in self.relations_by_field.items():
                 if not self.cleaned_data.get(field_name):
                     continue
                 for model_class in model_classes:
-                    self._clone_relations(model_class, instance)
+                    cloned_pairs = self._clone_relations(model_class, instance)
+                    cloned_pairs_by_model[model_class] = cloned_pairs
+
+        relink_cloned_codebase_resources(cloned_pairs_by_model)
 
         reevaluate_products([instance])
 

product_portfolio/tests/test_views.py

@@ -2187,14 +2187,18 @@ def test_product_portfolio_product_update_save_as_new(self):
         response = self.client.get(change_url)
         self.assertContains(response, expected)
 
-        ProductComponent.objects.create(
+        product_component = ProductComponent.objects.create(
             product=self.product1, component=self.component1, dataspace=self.dataspace
         )
-        ProductPackage.objects.create(
+        product_package = ProductPackage.objects.create(
             product=self.product1, package=self.package1, dataspace=self.dataspace
         )
         CodebaseResource.objects.create(
-            path="/path1/", product=self.product1, dataspace=self.dataspace
+            path="/path1/",
+            product=self.product1,
+            product_component=product_component,
+            product_package=product_package,
+            dataspace=self.dataspace,
         )
         ruleset = TriageRuleset.objects.create(
             name="Upgrade Ruleset", precedence=100, dataspace=self.dataspace
@@ -2225,6 +2229,11 @@ def test_product_portfolio_product_update_save_as_new(self):
         self.assertEqual(1, cloned_product.codebaseresources.count())
         self.assertEqual(1, cloned_product.product_triage_rulesets.count())
 
+        cloned_resource = cloned_product.codebaseresources.get()
+        self.assertEqual(cloned_product.productcomponents.get(), cloned_resource.product_component)
+        self.assertEqual(cloned_product.productpackages.get(), cloned_resource.product_package)
+        cloned_resource.full_clean()
+
     def test_product_portfolio_product_clone_view(self):
         clone_url = self.product1.get_clone_url()
         self.client.login(username=self.basic_user.username, password="secret")
@@ -2321,6 +2330,51 @@ def test_product_portfolio_product_clone_view(self):
         self.assertEqual(0, empty_clone.product_triage_rulesets.count())
         self.assertNotIn("view_product", get_user_perms(self.super_user, empty_clone))
 
+    def test_product_portfolio_product_clone_view_relinks_codebase_resources(self):
+        clone_url = self.product1.get_clone_url()
+        self.client.login(username=self.super_user.username, password="secret")
+
+        product_component = ProductComponent.objects.create(
+            product=self.product1, component=self.component1, dataspace=self.dataspace
+        )
+        product_package = ProductPackage.objects.create(
+            product=self.product1, package=self.package1, dataspace=self.dataspace
+        )
+        CodebaseResource.objects.create(
+            path="/path1/",
+            product=self.product1,
+            product_component=product_component,
+            product_package=product_package,
+            dataspace=self.dataspace,
+        )
+
+        data = {
+            "name": self.product1.name,
+            "version": "2.0",
+            "copy_inventory": "on",
+            "copy_codebase_resources": "on",
+        }
+        self.client.post(clone_url, data)
+        cloned_product = Product.unsecured_objects.get(name=self.product1.name, version="2.0")
+        cloned_resource = cloned_product.codebaseresources.get()
+        self.assertEqual(cloned_product.productcomponents.get(), cloned_resource.product_component)
+        self.assertEqual(cloned_product.productpackages.get(), cloned_resource.product_package)
+        cloned_resource.full_clean()
+
+        # Without the inventory, the links to the source product relations are cleared.
+        data["version"] = "3.0"
+        data["copy_inventory"] = ""
+        self.client.post(clone_url, data)
+        cloned_product = Product.unsecured_objects.get(name=self.product1.name, version="3.0")
+        cloned_resource = cloned_product.codebaseresources.get()
+        self.assertIsNone(cloned_resource.product_component)
+        self.assertIsNone(cloned_resource.product_package)
+        cloned_resource.full_clean()
+
+        source_resource = self.product1.codebaseresources.get()
+        self.assertEqual(product_component, source_resource.product_component)
+        self.assertEqual(product_package, source_resource.product_package)
+
     def test_product_portfolio_product_delete_view(self):
         delete_url = self.product1.get_delete_url()
         details_url = self.product1.get_absolute_url()