Add an overwrite boolean when modifying repo content

CHANGES/7550.feature

@@ -0,0 +1,3 @@
+Added an `overwrite` boolean parameter to the repository content modify and content upload
+endpoints. When set to `false`, the task will fail if the content being added would overwrite
+existing content based on `repo_key_fields`. Defaults to `true`.

pulp_file/tests/functional/api/test_overwrite_content.py

@@ -0,0 +1,225 @@
+"""Tests for the repository content overwrite protection feature."""
+
+import uuid
+
+import pytest
+
+from pulpcore.tests.functional.utils import PulpTaskError
+
+
+@pytest.mark.parallel
+def test_modify_overwrite_false_rejects_overwrite(
+    file_bindings,
+    file_repo,
+    file_content_unit_with_name_factory,
+    monitor_task,
+):
+    """Adding content with a conflicting relative_path and overwrite=False fails the task."""
+    shared_path = str(uuid.uuid4())
+
+    # Create two content units that share the same relative_path but have different artifacts
+    content_a = file_content_unit_with_name_factory(shared_path)
+    content_b = file_content_unit_with_name_factory(shared_path)
+    assert content_a.pulp_href != content_b.pulp_href
+
+    # Add the first content unit (overwrite is irrelevant for an empty repo)
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_a.pulp_href]},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # Attempt to add the second content unit with overwrite=False — expect task error
+    with pytest.raises(PulpTaskError) as exc_info:
+        monitor_task(
+            file_bindings.RepositoriesFileApi.modify(
+                file_repo.pulp_href,
+                {"add_content_units": [content_b.pulp_href], "overwrite": False},
+            ).task
+        )
+    assert exc_info.value.task.error["description"]
+    assert "PLP0023" in exc_info.value.task.error["description"]
+
+    # Repo version should not have advanced
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+
+@pytest.mark.parallel
+def test_modify_overwrite_false_allows_non_conflicting(
+    file_bindings,
+    file_repo,
+    file_content_unit_with_name_factory,
+    monitor_task,
+):
+    """Adding content without a conflicting relative_path and overwrite=False succeeds."""
+    content_a = file_content_unit_with_name_factory(str(uuid.uuid4()))
+    content_b = file_content_unit_with_name_factory(str(uuid.uuid4()))
+
+    # Add the first content unit
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_a.pulp_href], "overwrite": False},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # Add non-conflicting content with overwrite=False — should succeed
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_b.pulp_href], "overwrite": False},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/2/")
+
+
+@pytest.mark.parallel
+def test_modify_default_allows_overwrite(
+    file_bindings,
+    file_repo,
+    file_content_unit_with_name_factory,
+    monitor_task,
+):
+    """Default modify (overwrite=True) still allows overwriting content."""
+    shared_path = str(uuid.uuid4())
+    content_a = file_content_unit_with_name_factory(shared_path)
+    content_b = file_content_unit_with_name_factory(shared_path)
+
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_a.pulp_href]},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # With default overwrite=True, adding conflicting content succeeds
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_b.pulp_href]},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/2/")
+
+
+@pytest.mark.parallel
+def test_modify_overwrite_false_allows_replace_when_removing_conflict(
+    file_bindings,
+    file_repo,
+    file_content_unit_with_name_factory,
+    monitor_task,
+):
+    """Replacing content with overwrite=False succeeds when the conflicting unit is removed."""
+    shared_path = str(uuid.uuid4())
+    content_a = file_content_unit_with_name_factory(shared_path)
+    content_b = file_content_unit_with_name_factory(shared_path)
+
+    # Seed the repo with content_a
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {"add_content_units": [content_a.pulp_href]},
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # Remove content_a and add content_b in the same call with overwrite=False — should succeed
+    monitor_task(
+        file_bindings.RepositoriesFileApi.modify(
+            file_repo.pulp_href,
+            {
+                "remove_content_units": [content_a.pulp_href],
+                "add_content_units": [content_b.pulp_href],
+                "overwrite": False,
+            },
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/2/")
+
+
+@pytest.mark.parallel
+def test_content_upload_overwrite_false_rejects_overwrite(
+    file_bindings,
+    file_repo,
+    random_artifact_factory,
+    monitor_task,
+):
+    """Uploading content with overwrite=False and a conflicting relative_path is rejected."""
+    shared_path = str(uuid.uuid4())
+
+    # Upload first content unit into the repo
+    artifact_a = random_artifact_factory()
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            relative_path=shared_path,
+            artifact=artifact_a.pulp_href,
+            repository=file_repo.pulp_href,
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # Attempt to upload a second content unit with the same relative_path and overwrite=False
+    artifact_b = random_artifact_factory()
+    with pytest.raises(PulpTaskError) as exc_info:
+        monitor_task(
+            file_bindings.ContentFilesApi.create(
+                relative_path=shared_path,
+                artifact=artifact_b.pulp_href,
+                repository=file_repo.pulp_href,
+                overwrite=False,
+            ).task
+        )
+    assert exc_info.value.task.error["description"]
+    assert "PLP0023" in exc_info.value.task.error["description"]
+
+    # Repo version should not have advanced
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+
+@pytest.mark.parallel
+def test_content_upload_overwrite_false_allows_non_conflicting(
+    file_bindings,
+    file_repo,
+    random_artifact_factory,
+    monitor_task,
+):
+    """Uploading content with overwrite=False and no conflict succeeds."""
+    # Upload first content unit into the repo with overwrite=False
+    artifact_a = random_artifact_factory()
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            relative_path=str(uuid.uuid4()),
+            artifact=artifact_a.pulp_href,
+            repository=file_repo.pulp_href,
+            overwrite=False,
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/1/")
+
+    # Upload a second content unit with a different relative_path
+    artifact_b = random_artifact_factory()
+    monitor_task(
+        file_bindings.ContentFilesApi.create(
+            relative_path=str(uuid.uuid4()),
+            artifact=artifact_b.pulp_href,
+            repository=file_repo.pulp_href,
+            overwrite=False,
+        ).task
+    )
+    repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
+    assert repo.latest_version_href.endswith("/versions/2/")

pulpcore/app/models/repository.py

@@ -28,7 +28,7 @@
 )
 from pulpcore.constants import ALL_KNOWN_CONTENT_CHECKSUMS, PROTECTED_REPO_VERSION_MESSAGE
 from pulpcore.download.factory import DownloaderFactory
-from pulpcore.exceptions import ResourceImmutableError
+from pulpcore.exceptions import ContentOverwriteError, ResourceImmutableError
 
 from pulpcore.cache import Cache
 
@@ -242,6 +242,62 @@ def finalize_new_version(self, new_version):
         """
         pass
 
+    def check_content_overwrite(self, version, add_content_pks, remove_content_pks=None):
+        """
+        Check that content being added would not overwrite existing repository content.
+
+        Uses repo_key_fields to determine if any content to be added conflicts with content
+        already present in the version. Raises ContentOverwriteError (HTTP 409) if a conflict
+        is detected.
+
+        Plugin writers may override this method to customize overwrite-check behavior.
+
+        Args:
+            version (pulpcore.app.models.RepositoryVersion): The version to check against.
+            add_content_pks (list): List of primary keys for Content to be added.
+            remove_content_pks (list): Optional list of primary keys for Content being removed.
+                These are excluded from the conflict check.
+        """
+        existing_content = version.content
+        if remove_content_pks:
+            existing_content = existing_content.exclude(pk__in=remove_content_pks)
+
+        for type_obj in self.CONTENT_TYPES:
+            repo_key_fields = type_obj.repo_key_fields
+            if not repo_key_fields:
+                continue
+
+            for batch in batch_qs(
+                type_obj.objects.filter(pk__in=add_content_pks).values("pk", *repo_key_fields)
+            ):
+                find_dup_qs = Q()
+                incoming_by_key = {}
+                for content_dict in batch:
+                    pk = content_dict.pop("pk")
+                    key = tuple(content_dict[f] for f in repo_key_fields)
+                    incoming_by_key[key] = pk
+                    find_dup_qs |= Q(**content_dict)
+
+                existing_by_key = {
+                    tuple(row[f] for f in repo_key_fields): row["pk"]
+                    for row in type_obj.objects.filter(pk__in=existing_content)
+                    .filter(find_dup_qs)
+                    .values("pk", *repo_key_fields)
+                }
+                conflict_map = {
+                    incoming_by_key[key]: existing_by_key[key]
+                    for key in incoming_by_key
+                    if key in existing_by_key
+                }
+                if conflict_map:
+                    pulp_type = type_obj.get_pulp_type()
+                    _logger.info(
+                        "Content overwrite conflict: type=%s, incoming->existing=%s",
+                        pulp_type,
+                        conflict_map,
+                    )
+                    raise ContentOverwriteError(pulp_type, conflict_map)
+
     def latest_version(self):
         """
         Get the latest RepositoryVersion on a repository

pulpcore/app/serializers/content.py

@@ -33,6 +33,16 @@ class NoArtifactContentSerializer(base.ModelSerializer):
         view_name_pattern=r"repositories(-.*/.*)-detail",
         queryset=models.Repository.objects.all(),
     )
+    overwrite = serializers.BooleanField(
+        required=False,
+        default=True,
+        write_only=True,
+        help_text=_(
+            "When set to true, existing content in the repository with the same unique key "
+            "will be silently overwritten. When set to false, the task will fail if content "
+            "would be overwritten. Only used when 'repository' is specified. Defaults to true."
+        ),
+    )
     vuln_report = RelatedField(
         read_only=True,
         view_name="vuln_report-detail",
@@ -77,6 +87,7 @@ def create(self, validated_data):
             validated_data (dict): Data to save to the database
         """
         repository = validated_data.pop("repository", None)
+        overwrite = validated_data.pop("overwrite", True)
         artifacts = self.get_artifacts(validated_data)
 
         content = self.retrieve(validated_data)
@@ -113,9 +124,14 @@ def create(self, validated_data):
                 )
 
         if repository:
-            repository.cast()
+            repository = repository.cast()
             content_to_add = self.Meta.model.objects.filter(pk=content.pk)
 
+            if not overwrite:
+                version = repository.latest_version()
+                if version:
+                    repository.check_content_overwrite(version, [content.pk])
+
             # create new repo version with uploaded package
             with repository.new_version() as new_version:
                 new_version.add_content(content_to_add)
@@ -126,6 +142,7 @@ class Meta:
         model = models.Content
         fields = base.ModelSerializer.Meta.fields + (
             "repository",
+            "overwrite",
             "pulp_labels",
             "vuln_report",
         )

pulpcore/app/serializers/repository.py

@@ -333,6 +333,15 @@ class RepositoryAddRemoveContentSerializer(ModelSerializer, NestedHyperlinkedMod
             "for the new repository version"
         ),
     )
+    overwrite = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text=_(
+            "When set to true, existing content in the repository with the same unique key "
+            "will be silently overwritten. When set to false, the task will fail if content "
+            "would be overwritten. Defaults to true."
+        ),
+    )
 
     def validate_add_content_units(self, value):
         add_content_units = {}
@@ -366,4 +375,4 @@ def validate_remove_content_units(self, value):
 
     class Meta:
         model = models.RepositoryVersion
-        fields = ["add_content_units", "remove_content_units", "base_version"]
+        fields = ["add_content_units", "remove_content_units", "base_version", "overwrite"]