feat: add ability to clone a Product

dje/forms.py

@@ -45,6 +45,7 @@
 from dje.models import is_dataspace_related
 from dje.permissions import get_all_tabsets
 from dje.permissions import get_protected_fields
+from dje.utils import clone_related_objects
 from dje.utils import get_help_text
 from dje.utils import has_permission
 
@@ -210,21 +211,12 @@ def _clone_m2m(self, original_instance_id, cloned_instance):
         the `save_as_new` process.
         """
         field_name = self.instance._meta.model_name
+
         for model_class in self.clone_m2m_classes:
             if model_class.__name__ == "Subcomponent":
                 field_name = "parent"
 
-            related_instances = model_class.objects.filter(
-                **{
-                    f"{field_name}__id": original_instance_id,
-                }
-            )
-
-            for relation in related_instances:
-                relation.id = None
-                relation.uuid = uuid.uuid4()
-                setattr(relation, field_name, cloned_instance)
-                relation.save()
+            clone_related_objects(model_class, field_name, original_instance_id, cloned_instance)
 
     @property
     def save_as_new_submit(self):

dje/permissions.py

@@ -7,7 +7,9 @@
 #
 
 from guardian.shortcuts import assign_perm as guardian_assign_perm
+from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_perms_for_model
+from guardian.shortcuts import get_users_with_perms
 
 
 def get_protected_fields(model_class, user):
@@ -36,6 +38,19 @@ def assign_all_object_permissions(user, obj):
         guardian_assign_perm(perm, user, obj)
 
 
+def copy_object_permissions(source, target):
+    """Copy the user and group object permissions from `source` to `target`."""
+    users_with_perms = get_users_with_perms(source, attach_perms=True)
+    for user, perms in users_with_perms.items():
+        for perm in perms:
+            guardian_assign_perm(perm, user, target)
+
+    groups_with_perms = get_groups_with_perms(source, attach_perms=True)
+    for group, perms in groups_with_perms.items():
+        for perm in perms:
+            guardian_assign_perm(perm, group, target)
+
+
 def get_all_tabsets():
     """Return available tabs from all subclasses of ObjectDetailsView."""
     from dje.views import ObjectDetailsView

dje/tests/test_permissions.py

@@ -10,8 +10,13 @@
 from django.contrib.auth.models import Group
 from django.test import TestCase
 
+from guardian.shortcuts import assign_perm
+from guardian.shortcuts import get_group_perms
+from guardian.shortcuts import get_user_perms
+
 from dje.models import Dataspace
 from dje.models import DataspaceConfiguration
+from dje.permissions import copy_object_permissions
 from dje.permissions import get_all_tabsets
 from dje.permissions import get_authorized_tabs
 from dje.permissions import get_protected_fields
@@ -237,3 +242,26 @@ def test_permissions_get_authorized_tabs(self):
 
         # Superuser users see all the tabs
         self.assertIsNone(get_authorized_tabs(Owner, self.super_user))
+
+    def test_permissions_copy_object_permissions(self):
+        Product = apps.get_model("product_portfolio", "Product")
+        source = Product.objects.create(name="Source", dataspace=self.dataspace)
+        target = Product.objects.create(name="Target", dataspace=self.dataspace)
+        group1 = Group.objects.create(name="Group1")
+
+        assign_perm("view_product", self.basic_user, source)
+        assign_perm("change_product", self.basic_user, source)
+        assign_perm("view_product", group1, source)
+
+        self.assertEqual(set(), set(get_user_perms(self.basic_user, target)))
+        self.assertEqual(set(), set(get_group_perms(group1, target)))
+
+        copy_object_permissions(source, target)
+
+        expected_user_perms = {"view_product", "change_product"}
+        self.assertEqual(expected_user_perms, set(get_user_perms(self.basic_user, target)))
+        self.assertEqual({"view_product"}, set(get_group_perms(group1, target)))
+
+        # The source keeps its own permissions untouched.
+        self.assertEqual(expected_user_perms, set(get_user_perms(self.basic_user, source)))
+        self.assertEqual({"view_product"}, set(get_group_perms(group1, source)))

dje/tests/test_utils.py

@@ -25,6 +25,7 @@
 from dejacode_toolkit.utils import sha1
 from dje.copier import copy_object
 from dje.models import Dataspace
+from dje.utils import clone_related_objects
 from dje.utils import database_re_escape
 from dje.utils import extract_name_version
 from dje.utils import get_duplicates
@@ -54,6 +55,8 @@
 License = apps.get_model("license_library", "license")
 ExternalReference = apps.get_model("dje", "ExternalReference")
 ExternalSource = apps.get_model("dje", "ExternalSource")
+Product = apps.get_model("product_portfolio", "Product")
+ProductComponent = apps.get_model("product_portfolio", "ProductComponent")
 
 
 class DJEUtilsTestCase(TestCase):
@@ -227,6 +230,27 @@ def test_dje_utils_merge_relations(self):
         with self.assertRaises(AssertionError):
             merge_relations(original, alternate_owner)
 
+    def test_dje_utils_clone_related_objects(self):
+        nexb_dataspace = Dataspace.objects.create(name="nexB")
+        product1 = Product.objects.create(name="p1", dataspace=nexb_dataspace)
+        product2 = Product.objects.create(name="p2", dataspace=nexb_dataspace)
+        component1 = Component.objects.create(name="c1", dataspace=nexb_dataspace)
+        relation1 = ProductComponent.objects.create(
+            product=product1, component=component1, dataspace=nexb_dataspace
+        )
+
+        cloned_pairs = clone_related_objects(ProductComponent, "product", product1.id, product2)
+
+        self.assertEqual(1, product1.productcomponents.count())
+        self.assertEqual(relation1, product1.productcomponents.get())
+        cloned_relation = product2.productcomponents.get()
+        self.assertNotEqual(relation1.pk, cloned_relation.pk)
+        self.assertNotEqual(relation1.uuid, cloned_relation.uuid)
+        self.assertEqual(component1, cloned_relation.component)
+        self.assertEqual(product2.dataspace, cloned_relation.dataspace)
+
+        self.assertEqual([(relation1.pk, cloned_relation)], cloned_pairs)
+
     def test_dje_utils_group_by_name_version(self):
         test_cases = [
             {

dje/utils.py

@@ -336,6 +336,27 @@ def get_model_class_from_path(path):
     return apps.get_model(app_name, model)
 
 
+def clone_related_objects(model_class, fk_field_name, source_id, target_object, save_kwargs=None):
+    """
+    Duplicate the `model_class` instances related to `source_id` onto `target_object`.
+
+    Return a list of (original_pk, cloned_instance) pairs.
+    """
+    related_objects = model_class.objects.filter(**{f"{fk_field_name}__id": source_id})
+    save_kwargs = save_kwargs or {}
+
+    cloned_pairs = []
+    for relation in related_objects:
+        original_pk = relation.pk
+        relation.id = None
+        relation.uuid = uuid.uuid4()
+        setattr(relation, fk_field_name, target_object)
+        relation.save(**save_kwargs)
+        cloned_pairs.append((original_pk, relation))
+
+    return cloned_pairs
+
+
 def merge_relations(original, duplicate):
     """Move `original` object references (ManyToOneRel, GenericRelation) from `duplicate`."""
     if original.__class__ != duplicate.__class__ or original.dataspace != duplicate.dataspace:

product_portfolio/api.py

@@ -44,6 +44,7 @@
 from product_portfolio.forms import ImportFromScanForm
 from product_portfolio.forms import ImportManifestsForm
 from product_portfolio.forms import LoadSBOMsForm
+from product_portfolio.forms import ProductCloneForm
 from product_portfolio.forms import PullProjectDataForm
 from product_portfolio.models import CodebaseResource
 from product_portfolio.models import Product
@@ -407,6 +408,45 @@ class AssignTriageRulesetSerializer(serializers.Serializer):
     assigned = serializers.BooleanField()
 
 
+class CanChangeProduct(permissions.BasePermission):
+    """Allows the action only if the user has the `change_product` object permission."""
+
+    def has_object_permission(self, request, view, obj):
+        return request.user.has_perm("product_portfolio.change_product", obj)
+
+
+class ProductCloneSerializer(serializers.Serializer):
+    name = serializers.CharField(
+        required=True,
+        help_text=ProductCloneForm.base_fields["name"].help_text,
+    )
+    version = serializers.CharField(
+        required=False,
+        allow_blank=True,
+        help_text=ProductCloneForm.base_fields["version"].help_text,
+    )
+    copy_inventory = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text=str(ProductCloneForm.base_fields["copy_inventory"].label),
+    )
+    copy_codebase_resources = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text=str(ProductCloneForm.base_fields["copy_codebase_resources"].label),
+    )
+    copy_triage_rulesets = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text=str(ProductCloneForm.base_fields["copy_triage_rulesets"].label),
+    )
+    copy_object_permissions = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text=str(ProductCloneForm.base_fields["copy_object_permissions"].label),
+    )
+
+
 class ProductViewSet(
     ObjectPermissionsMixin,
     SendAboutFilesMixin,
@@ -465,6 +505,43 @@ def perform_create(self, serializer):
         super().perform_create(serializer)
         assign_all_object_permissions(self.request.user, serializer.instance)
 
+    @action(
+        detail=True,
+        methods=["post"],
+        serializer_class=ProductCloneSerializer,
+        permission_classes=[
+            permissions.IsAuthenticated,
+            permissions.DjangoModelPermissions,
+            CanChangeProduct,
+        ],
+    )
+    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,
+        depending on the submitted flags (all default to true).
+        """
+        product = self.get_object()
+
+        clone_serializer = ProductCloneSerializer(data=request.data)
+        if not clone_serializer.is_valid():
+            return Response(clone_serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        form = ProductCloneForm(
+            user=request.user,
+            source_product=product,
+            data=clone_serializer.validated_data,
+        )
+        if not form.is_valid():
+            return Response(form.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        cloned_product = form.save()
+        serializer = ProductSerializer(cloned_product, context=self.get_serializer_context())
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
     @action(detail=True)
     def imports(self, request, uuid):
         """