From cfb572660dfd823181e0867454a79e351ca39dfb Mon Sep 17 00:00:00 2001 From: tdruez <489057+tdruez@users.noreply.github.com> Date: Mon, 27 May 2024 11:48:53 +0400 Subject: [PATCH] Add replace_existing_version field on the AddToProduct form #12 (#124) Signed-off-by: tdruez --- CHANGELOG.rst | 4 + component_catalog/forms.py | 19 ++- component_catalog/models.py | 6 +- .../component_catalog/add_to_product.html | 9 +- .../component_catalog/includes/add_to.js.html | 48 ++++-- .../includes/add_to_modal.html | 4 +- component_catalog/tests/test_models.py | 3 + component_catalog/views.py | 15 +- dje/models.py | 13 +- policy/models.py | 2 +- product_portfolio/models.py | 121 +++++++++---- product_portfolio/tests/test_models.py | 160 +++++++++++++++++- 12 files changed, 342 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index ee11fe2e..fb6b2937 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -78,6 +78,10 @@ Release notes - Enhance Package Import to support modifications. https://github.com/nexB/dejacode/issues/84 +- Add an option on the "Add to Product" form to to replace any existing relationships + with a different version of the same object by the selected object. + https://github.com/nexB/dejacode/issues/12 + ### Version 5.0.1 - Improve the stability of the "Check for new Package versions" feature. diff --git a/component_catalog/forms.py b/component_catalog/forms.py index 73a4cb60..e6c3ddf8 100644 --- a/component_catalog/forms.py +++ b/component_catalog/forms.py @@ -646,6 +646,17 @@ class AddToProductAdminForm(forms.Form): queryset=Product.objects.none(), ) ids = forms.CharField(widget=forms.widgets.HiddenInput) + replace_existing_version = forms.BooleanField( + required=False, + initial=False, + label="Replace existing relationships by newer version.", + help_text=( + "Select this option to replace any existing relationships with a different version " + "of the same object. " + "If more than one version of the object is already assigned, no replacements will be " + "made, and the new version will be added instead." + ), + ) def __init__(self, request, model, relation_model, *args, **kwargs): super().__init__(*args, **kwargs) @@ -663,9 +674,11 @@ def get_selected_objects(self): def save(self): product = self.cleaned_data["product"] + return product.assign_objects( related_objects=self.get_selected_objects(), user=self.request.user, + replace_version=self.cleaned_data["replace_existing_version"], ) @@ -719,15 +732,15 @@ def new_component_from_package_link(self): href = f"{component_add_url}?package_ids={package.id}" return HTML( + f"
" f'
' f' ' f" Add Component from Package data" f" " f"
" - f"
" ) def clean_component(self): @@ -749,9 +762,9 @@ def helper(self): helper.layout = Layout( Fieldset( None, - self.new_component_from_package_link(), "object_id", "component", + self.new_component_from_package_link(), ), ) return helper diff --git a/component_catalog/models.py b/component_catalog/models.py index a220135b..e4917f4c 100644 --- a/component_catalog/models.py +++ b/component_catalog/models.py @@ -1915,12 +1915,16 @@ def get_export_cyclonedx_url(self): return self.get_url("export_cyclonedx") @classmethod - def get_identifier_fields(cls): + def get_identifier_fields(cls, *args, purl_fields_only=False, **kwargs): """ Explicit list of identifier fields as we do not enforce a unique together on this model. This is used in the Importer, to catch duplicate entries. + The purl_fields_only option can be use to limit the results. """ + if purl_fields_only: + return PACKAGE_URL_FIELDS + return ["filename", "download_url", *PACKAGE_URL_FIELDS] @property diff --git a/component_catalog/templates/admin/component_catalog/add_to_product.html b/component_catalog/templates/admin/component_catalog/add_to_product.html index 854add47..a1621436 100644 --- a/component_catalog/templates/admin/component_catalog/add_to_product.html +++ b/component_catalog/templates/admin/component_catalog/add_to_product.html @@ -19,17 +19,22 @@

Add {% trans opts.verbose_name %} to a product

The Product provides you with a group of {% trans opts.verbose_name_plural|title %} that are used together, so that you can generate Attribution documentation for all of the {% trans opts.verbose_name_plural|title %} in that Product.

-

+

{{ form.non_field_errors }} {{ form.product.errors }} {{ form.product }} +
+ {{ form.replace_existing_version }} + {{ form.replace_existing_version.label }} +

{{ form.replace_existing_version.help_text }}

+
{% if perms.product_portfolio.add_product %} {% endif %} {{ form.ct }} {{ form.ids }} -

+

diff --git a/component_catalog/templates/component_catalog/includes/add_to.js.html b/component_catalog/templates/component_catalog/includes/add_to.js.html index 77e2ffeb..a024fe67 100644 --- a/component_catalog/templates/component_catalog/includes/add_to.js.html +++ b/component_catalog/templates/component_catalog/includes/add_to.js.html @@ -1,29 +1,35 @@ \ No newline at end of file + \ No newline at end of file diff --git a/component_catalog/templates/component_catalog/includes/add_to_modal.html b/component_catalog/templates/component_catalog/includes/add_to_modal.html index 122323ce..847f1758 100644 --- a/component_catalog/templates/component_catalog/includes/add_to_modal.html +++ b/component_catalog/templates/component_catalog/includes/add_to_modal.html @@ -8,13 +8,13 @@
{{ form.helper.modal_title }}
+ {% crispy form %} {# Only displayed on list views #} {% if request.resolver_match.url_name|default:""|slice:"-4:" == "list" %} +
Selected objects:
    -
    {% endif %} - {% crispy form %}
    diff --git a/component_catalog/tests/test_models.py b/component_catalog/tests/test_models.py index b03e5509..03d5ced0 100644 --- a/component_catalog/tests/test_models.py +++ b/component_catalog/tests/test_models.py @@ -26,6 +26,7 @@ from component_catalog.importers import ComponentImporter from component_catalog.importers import PackageImporter +from component_catalog.models import PACKAGE_URL_FIELDS from component_catalog.models import Component from component_catalog.models import ComponentAssignedLicense from component_catalog.models import ComponentAssignedPackage @@ -369,6 +370,8 @@ def test_component_catalog_models_get_identifier_fields(self): for model_class, expected in inputs: self.assertEqual(expected, model_class.get_identifier_fields()) + self.assertEqual(PACKAGE_URL_FIELDS, Package.get_identifier_fields(purl_fields_only=True)) + def test_component_model_get_absolute_url(self): c = Component(name="c1", version="1.0", dataspace=self.dataspace) self.assertEqual("/components/nexB/c1/1.0/", c.get_absolute_url()) diff --git a/component_catalog/views.py b/component_catalog/views.py index a37cff59..abb89fe9 100644 --- a/component_catalog/views.py +++ b/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,12 +942,16 @@ 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}." + if updated_count: + msg += f" {updated_count} {model_name}(s) updated on {product_name}." if unchanged_count: msg += f" {unchanged_count} {model_name}(s) were already assigned." + messages.success(self.request, msg) return super().form_valid(form) diff --git a/dje/models.py b/dje/models.py index 3112dd83..794e811d 100644 --- a/dje/models.py +++ b/dje/models.py @@ -773,6 +773,17 @@ def update_from_data(self, user, data, override=False): return updated_fields + def update(self, **kwargs): + """ + Update this instance with the provided ``kwargs`` values. + The full ``save()`` process will be triggered, including signals, and the + ``update_fields`` is automatically set. + """ + for field_name, value in kwargs.items(): + setattr(self, field_name, value) + + self.save(update_fields=list(kwargs.keys())) + def as_json(self): try: serialized_data = serialize( @@ -881,7 +892,7 @@ def _get_local_foreign_fields(self): local_foreign_fields = property(_get_local_foreign_fields) @classmethod - def get_identifier_fields(cls): + def get_identifier_fields(cls, *args, **kwargs): """ Return a list of the fields, based on the Meta unique_together, to be used to match a unique instance within a Dataspace. diff --git a/policy/models.py b/policy/models.py index 4fd78498..e76155cb 100755 --- a/policy/models.py +++ b/policy/models.py @@ -107,7 +107,7 @@ def str_with_content_type(self): return f"{self.label} ({self.content_type.model})" @classmethod - def get_identifier_fields(cls): + def get_identifier_fields(cls, *args, **kwargs): """Hack required by the Component import.""" return ["label"] diff --git a/product_portfolio/models.py b/product_portfolio/models.py index e33d83f5..b14047bd 100644 --- a/product_portfolio/models.py +++ b/product_portfolio/models.py @@ -332,49 +332,110 @@ def get_cyclonedx_components(self): *list(self.productpackages.all().order_by("id")), ] - def assign_objects(self, related_objects, user): - """ - Assign provided `related_objects` to this `Product`. - Supported object models are `Component` and `Package`. - Return the both counts for created and unchanged objects. - """ - created_count = 0 - unchanged_count = 0 - + def get_relationship_model(self, obj): + """Return the relationship model class for a given object.""" relationship_models = { "component": ProductComponent, "package": ProductPackage, } + object_model_name = obj._meta.model_name # "component" or "package" + relationship_model = relationship_models.get(object_model_name) + if not relationship_model: + raise ValueError(f"Unsupported object model: {object_model_name}") + + return relationship_model + + def find_assigned_other_versions(self, obj): + """ + Look for the same objects with a different version already assigned to the product. + Return the relation queryset for the objects with a different version. + """ + object_model_name = obj._meta.model_name # "component" or "package" + relationship_model = self.get_relationship_model(obj) + + # Craft the lookups excluding the version field + no_version_object_lookups = { + f"{object_model_name}__{field_name}": getattr(obj, field_name) + for field_name in obj.get_identifier_fields(purl_fields_only=True) + if field_name != "version" + } + + filters = { + "product": self, + "dataspace": obj.dataspace, + **no_version_object_lookups, + } + excludes = { + f"{object_model_name}__id": obj.id, + } + + return relationship_model.objects.exclude(**excludes).filter(**filters) + + def assign_object(self, obj, user, replace_version=False): + """ + Assign a provided ``obj`` (either a Component or Package) to this Product. + 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) + + filters = { + "product": self, + "dataspace": obj.dataspace, + object_model_name: obj, + } + + # 1. Check if a relation for this object already exists + if relationship_model.objects.filter(**filters).exists(): + 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: + 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 "updated", existing_relation + + # 3. Otherwise, create a new relation + defaults = { + "license_expression": obj.license_expression, + "created_by": user, + "last_modified_by": user, + } + 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", created_relation + + def assign_objects(self, related_objects, user, replace_version=False): + """ + 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: - relation_model_name = obj._meta.model_name # 'component' or 'package' - relation_model_class = relationship_models.get(relation_model_name) - if not relation_model_class: - continue - - filters = { - "product": self, - relation_model_name: obj, - "dataspace": obj.dataspace, - "defaults": { - "license_expression": obj.license_expression, - "created_by": user, - "last_modified_by": user, - }, - } - relation_obj, created = relation_model_class.objects.get_or_create(**filters) - if created: - History.log_addition(user, relation_obj) - History.log_change(user, self, f'Added {relation_model_name} "{obj}"') + 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: + 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): """ diff --git a/product_portfolio/tests/test_models.py b/product_portfolio/tests/test_models.py index 56cb335f..1149e476 100644 --- a/product_portfolio/tests/test_models.py +++ b/product_portfolio/tests/test_models.py @@ -248,8 +248,13 @@ def test_product_model_assign_objects(self): ) self.assertEqual(0, self.product1.productcomponents.count()) - created, unchanged = self.product1.assign_objects([], self.super_user) + with self.assertRaises(ValueError) as cm: + self.product1.assign_objects([status1], self.super_user) + self.assertEqual("Unsupported object model: productrelationstatus", str(cm.exception)) + + 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( @@ -257,8 +262,9 @@ def test_product_model_assign_objects(self): 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) @@ -281,6 +287,156 @@ def test_product_model_assign_objects(self): ) self.assertEqual(self.super_user, self.product1.last_modified_by) + 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) + + # 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()) + + # 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() + 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) + + def test_product_model_assign_object_replace_version_package(self): + package_data = { + "filename": "package.zip", + "type": "deb", + "namespace": "debian", + "name": "curl", + "dataspace": self.dataspace, + } + package1 = Package.objects.create(**package_data, version="1.0") + package2 = Package.objects.create(**package_data, version="2.0") + + # 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()) + # 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() + 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) + + def test_product_model_find_assigned_other_versions_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) + component3 = Component.objects.create(name="c", version="3.0", dataspace=self.dataspace) + + # No other version assigned + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(component1)) + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(component2)) + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(component3)) + + # 1 other version assigned + _, 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) + self.assertQuerySetEqual([p1_c2], self.product1.find_assigned_other_versions(component1)) + self.assertQuerySetEqual([p1_c1], self.product1.find_assigned_other_versions(component2)) + self.assertQuerySetEqual( + [p1_c1, p1_c2], self.product1.find_assigned_other_versions(component3) + ) + + def test_product_model_find_assigned_other_versions_package(self): + package_data = { + "filename": "package.zip", + "type": "deb", + "namespace": "debian", + "name": "curl", + "dataspace": self.dataspace, + } + package1 = Package.objects.create(**package_data, version="1.0") + package2 = Package.objects.create(**package_data, version="2.0") + package3 = Package.objects.create(**package_data, version="3.0") + + # No other version assigned + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(package1)) + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(package2)) + self.assertQuerySetEqual([], self.product1.find_assigned_other_versions(package3)) + + # 1 other version assigned + _, 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) + self.assertQuerySetEqual([p1_p2], self.product1.find_assigned_other_versions(package1)) + self.assertQuerySetEqual([p1_p1], self.product1.find_assigned_other_versions(package2)) + self.assertQuerySetEqual( + [p1_p1, p1_p2], self.product1.find_assigned_other_versions(package3) + ) + + # Only PURL fields are used as lookups as the filename and download_url + # fields change between version. + package_data["filename"] = "different_filename" + package4 = Package.objects.create(**package_data, version="4.0") + self.assertQuerySetEqual( + [p1_p1, p1_p2], self.product1.find_assigned_other_versions(package4) + ) + def test_product_model_field_changes_mixin(self): self.assertFalse(Product().has_changed("name"))