Refine assign_objects with a new updated status #12

Signed-off-by: tdruez <tdruez@nexb.com>

component_catalog/views.py

@@ -217,12 +217,17 @@ def get_form_kwargs(self):
         return kwargs
 
     def form_valid(self, form):
-        created_count, unchanged_count = form.save()
+        created_count, updated_count, unchanged_count = form.save()
         product = form.cleaned_data["product"]
         opts = self.model._meta
 
+        msg = ""
         if created_count:
-            msg = f'{created_count} {opts.model_name}(s) added to "{product}".'
+            msg = f'{created_count} {opts.model_name}(s) added to "{product}". '
+        if updated_count:
+            msg += f'{updated_count} {opts.model_name}(s) updated on "{product}".'
+
+        if msg:
             messages.success(self.request, msg)
         else:
             msg = f"No new {opts.model_name}(s) were assigned to this product."
@@ -937,7 +942,7 @@ def get_form_kwargs(self):
         return kwargs
 
     def form_valid(self, form):
-        created_count, unchanged_count = form.save()
+        created_count, updated_count, unchanged_count = form.save()
         model_name = self.model._meta.model_name
         product_name = form.cleaned_data["product"].name
         msg = f"{created_count} {model_name}(s) added to {product_name}."

product_portfolio/models.py

@@ -374,7 +374,8 @@ def find_assigned_other_versions(self, obj):
     def assign_object(self, obj, user, replace_version=False):
         """
         Assign a provided ``obj`` (either a Component or Package) to this Product.
-        Return the created or existing relationship object.
+        Return a tuple with the status ("created", "updated", "unchanged") and the relationship
+        object.
         """
         object_model_name = obj._meta.model_name  # "component" or "package"
         relationship_model = self.get_relationship_model(obj)
@@ -387,19 +388,19 @@ def assign_object(self, obj, user, replace_version=False):
 
         # 1. Check if a relation for this object already exists
         if relationship_model.objects.filter(**filters).exists():
-            return
+            return "unchanged", None
 
         # 2. Find an existing relation for another version of the same object
         #    when replace_version is provided.
         if replace_version:
             other_assigned_versions = self.find_assigned_other_versions(obj)
             if len(other_assigned_versions) == 1:
-                exsisting_relation = other_assigned_versions[0]
-                other_version_object = getattr(exsisting_relation, object_model_name)
-                exsisting_relation.update(**{object_model_name: obj, "last_modified_by": user})
+                existing_relation = other_assigned_versions[0]
+                other_version_object = getattr(existing_relation, object_model_name)
+                existing_relation.update(**{object_model_name: obj, "last_modified_by": user})
                 message = f'Updated {object_model_name} "{other_version_object}" to "{obj}"'
                 History.log_change(user, self, message)
-                return exsisting_relation
+                return "updated", existing_relation
 
         # 3. Otherwise, create a new relation
         defaults = {
@@ -410,25 +411,31 @@ def assign_object(self, obj, user, replace_version=False):
         created_relation = relationship_model.objects.create(**filters, **defaults)
         History.log_addition(user, created_relation)
         History.log_change(user, self, f'Added {object_model_name} "{obj}"')
-        return created_relation
+        return "created", created_relation
 
     def assign_objects(self, related_objects, user, replace_version=False):
-        """Assign provided ``related_objects`` (either a Component or Package) to this Product."""
+        """
+        Assign provided ``related_objects`` (either a Component or Package) to this Product.
+        Return counts of created, updated, and unchanged objects.
+        """
         created_count = 0
+        updated_count = 0
         unchanged_count = 0
 
         for obj in related_objects:
-            created_relation = self.assign_object(obj, user, replace_version)
-            if created_relation:
+            status, relation = self.assign_object(obj, user, replace_version)
+            if status == "created":
                 created_count += 1
+            elif status == "updated":
+                updated_count += 1
             else:
                 unchanged_count += 1
 
-        if created_count > 0:
+        if created_count > 0 or updated_count > 0:
             self.last_modified_by = user
             self.save()
 
-        return created_count, unchanged_count
+        return created_count, updated_count, unchanged_count
 
     def scan_all_packages_task(self, user):
         """

product_portfolio/tests/test_models.py

@@ -252,17 +252,19 @@ def test_product_model_assign_objects(self):
             self.product1.assign_objects([status1], self.super_user)
         self.assertEqual("Unsupported object model: productrelationstatus", str(cm.exception))
 
-        created, unchanged = self.product1.assign_objects([], self.super_user)
+        created, updated, unchanged = self.product1.assign_objects([], self.super_user)
         self.assertEqual(0, created)
+        self.assertEqual(0, updated)
         self.assertEqual(0, unchanged)
 
         component1 = Component.objects.create(
             name="c1",
             license_expression=self.license1.key,
             dataspace=self.dataspace,
         )
-        created, unchanged = self.product1.assign_objects([component1], self.super_user)
+        created, updated, unchanged = self.product1.assign_objects([component1], self.super_user)
         self.assertEqual(1, created)
+        self.assertEqual(0, updated)
         self.assertEqual(0, unchanged)
 
         pc = ProductComponent.objects.get(product=self.product1, component=component1)
@@ -288,26 +290,44 @@ def test_product_model_assign_objects(self):
     def test_product_model_assign_object_replace_version_component(self):
         component1 = Component.objects.create(name="c", version="1.0", dataspace=self.dataspace)
         component2 = Component.objects.create(name="c", version="2.0", dataspace=self.dataspace)
-        self.product1.assign_object(component1, self.super_user)
+
+        # Assign component1
+        status, relation = self.product1.assign_object(component1, self.super_user)
+        self.assertEqual("created", status)
+        self.assertTrue(relation)
+        self.assertQuerySetEqual([component1], self.product1.components.all())
+
+        # Re-assign component1, relation already exists.
+        status, relation = self.product1.assign_object(component1, self.super_user)
+        self.assertEqual("unchanged", status)
+        self.assertIsNone(relation)
+        self.assertQuerySetEqual([component1], self.product1.components.all())
+        # With replace_version
+        status, relation = self.product1.assign_object(
+            component1, self.super_user, replace_version=1
+        )
+        self.assertEqual("unchanged", status)
+        self.assertIsNone(relation)
         self.assertQuerySetEqual([component1], self.product1.components.all())
-        p1_c2 = self.product1.assign_object(component2, self.super_user)
+
+        # Assign component2
+        status, p1_c2 = self.product1.assign_object(component2, self.super_user)
+        self.assertEqual("created", status)
+        self.assertTrue(p1_c2)
         self.assertQuerySetEqual([component1, component2], self.product1.components.all())
 
         # Replacing the current single existing version.
         p1_c2.delete()
-        p1_c2 = self.product1.assign_object(component2, self.super_user, replace_version=True)
+        status, p1_c2 = self.product1.assign_object(
+            component2, self.super_user, replace_version=True
+        )
+        self.assertEqual("updated", status)
+        self.assertTrue(p1_c2)
         self.assertQuerySetEqual([component2], self.product1.components.all())
 
         history_entries = History.objects.get_for_object(self.product1)
         expected_message = 'Updated component "c 1.0" to "c 2.0"'
         self.assertEqual(expected_message, history_entries.latest("action_time").change_message)
-        self.product1.refresh_from_db()
-        self.assertEqual(self.super_user, self.product1.last_modified_by)
-
-        # Relation already exists.
-        self.assertIsNone(
-            self.product1.assign_object(component2, self.super_user, replace_version=True)
-        )
 
     def test_product_model_assign_object_replace_version_package(self):
         package_data = {
@@ -320,26 +340,39 @@ def test_product_model_assign_object_replace_version_package(self):
         package1 = Package.objects.create(**package_data, version="1.0")
         package2 = Package.objects.create(**package_data, version="2.0")
 
-        self.product1.assign_object(package1, self.super_user)
+        # Assign package1
+        status, relation = self.product1.assign_object(package1, self.super_user)
+        self.assertEqual("created", status)
+        self.assertTrue(relation)
+        self.assertQuerySetEqual([package1], self.product1.packages.all())
+
+        # Re-assign package1, relation already exists.
+        status, relation = self.product1.assign_object(package1, self.super_user)
+        self.assertEqual("unchanged", status)
+        self.assertIsNone(relation)
         self.assertQuerySetEqual([package1], self.product1.packages.all())
-        p1_p2 = self.product1.assign_object(package2, self.super_user)
+        # With replace_version
+        status, relation = self.product1.assign_object(package1, self.super_user, replace_version=1)
+        self.assertEqual("unchanged", status)
+        self.assertIsNone(relation)
+        self.assertQuerySetEqual([package1], self.product1.packages.all())
+
+        # Assign package2
+        status, p1_p2 = self.product1.assign_object(package2, self.super_user)
+        self.assertEqual("created", status)
+        self.assertTrue(p1_p2)
         self.assertQuerySetEqual([package1, package2], self.product1.packages.all())
 
         # Replacing the current single existing version.
         p1_p2.delete()
-        p1_p2 = self.product1.assign_object(package2, self.super_user, replace_version=True)
+        status, p1_p2 = self.product1.assign_object(package2, self.super_user, replace_version=True)
+        self.assertEqual("updated", status)
+        self.assertTrue(p1_p2)
         self.assertQuerySetEqual([package2], self.product1.packages.all())
 
         history_entries = History.objects.get_for_object(self.product1)
         expected_message = 'Updated package "pkg:deb/debian/curl@1.0" to "pkg:deb/debian/curl@2.0"'
         self.assertEqual(expected_message, history_entries.latest("action_time").change_message)
-        self.product1.refresh_from_db()
-        self.assertEqual(self.super_user, self.product1.last_modified_by)
-
-        # Relation already exists.
-        self.assertIsNone(
-            self.product1.assign_object(package2, self.super_user, replace_version=True)
-        )
 
     def test_product_model_find_assigned_other_versions_component(self):
         component1 = Component.objects.create(name="c", version="1.0", dataspace=self.dataspace)
@@ -352,13 +385,13 @@ def test_product_model_find_assigned_other_versions_component(self):
         self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(component3))
 
         # 1 other version assigned
-        p1_c1 = self.product1.assign_object(component1, self.super_user)
+        _, p1_c1 = self.product1.assign_object(component1, self.super_user)
         self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(component1))
         self.assertQuerySetEqual([p1_c1], self.product1.find_assigned_other_versions(component2))
         self.assertQuerySetEqual([p1_c1], self.product1.find_assigned_other_versions(component3))
 
         # 2 other versions assigned
-        p1_c2 = self.product1.assign_object(component2, self.super_user)
+        _, p1_c2 = self.product1.assign_object(component2, self.super_user)
         self.assertQuerySetEqual([p1_c2], self.product1.find_assigned_other_versions(component1))
         self.assertQuerySetEqual([p1_c1], self.product1.find_assigned_other_versions(component2))
         self.assertQuerySetEqual(
@@ -383,13 +416,13 @@ def test_product_model_find_assigned_other_versions_package(self):
         self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(package3))
 
         # 1 other version assigned
-        p1_p1 = self.product1.assign_object(package1, self.super_user)
+        _, p1_p1 = self.product1.assign_object(package1, self.super_user)
         self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(package1))
         self.assertQuerySetEqual([p1_p1], self.product1.find_assigned_other_versions(package2))
         self.assertQuerySetEqual([p1_p1], self.product1.find_assigned_other_versions(package3))
 
         # 2 other versions assigned
-        p1_p2 = self.product1.assign_object(package2, self.super_user)
+        _, p1_p2 = self.product1.assign_object(package2, self.super_user)
         self.assertQuerySetEqual([p1_p2], self.product1.find_assigned_other_versions(package1))
         self.assertQuerySetEqual([p1_p1], self.product1.find_assigned_other_versions(package2))
         self.assertQuerySetEqual(