Refactor CollectionVersion Upload to use pulpcore machinery

CHANGES/1175.bugfix

@@ -0,0 +1 @@
+Properly return 400 error when trying to create/upload a duplicate Collection.

CHANGES/1175.feature

@@ -0,0 +1 @@
+An existing artifact or upload object can now be used to create a Collection.

CHANGES/1176.removal

@@ -0,0 +1,5 @@
+Renamed CollectionVersion upload fields [namespace, name, version] to expected_[namespace, name, version].
+
+Deprecated /ansible/collections/ upload endpoint. Use /pulp/api/v3/content/ansible/collection_versions/ instead.
+
+Deprecated Galaxy V2 Collection upload endpoint. Use Galaxy V3 Collection Artifact upload endpoint instead.

pulp_ansible/app/galaxy/mixins.py

@@ -19,3 +19,9 @@ def _dispatch_import_collection_task(self, temp_file_pk, repository=None, **kwar
             kwargs["repository_pk"] = repository.pk
 
         return dispatch(import_collection, exclusive_resources=locks, kwargs=kwargs)
+
+    def get_deferred_context(self, request):
+        context = {}
+        if "file" in request.data:
+            context["filename"] = request.data["file"].name
+        return context

pulp_ansible/app/galaxy/serializers.py

@@ -6,6 +6,7 @@
 from rest_framework.reverse import reverse
 from rest_framework import serializers
 
+from pulpcore.plugin.models import Artifact
 from pulp_ansible.app.models import Collection, CollectionVersion, Role
 from pulp_ansible.app.galaxy.v3.serializers import CollectionMetadataSerializer
 
@@ -187,3 +188,10 @@ class GalaxyCollectionUploadSerializer(serializers.Serializer):
     file = serializers.FileField(
         help_text=_("The file containing the Artifact binary data."), required=True
     )
+
+    def validate(self, data):
+        """Ensure duplicate artifact isn't uploaded."""
+        sha256 = data["file"].hashers["sha256"].hexdigest()
+        artifact = Artifact.objects.filter(sha256=sha256).first()
+        if artifact:
+            raise serializers.ValidationError(_("Artifact already exists"))

pulp_ansible/app/galaxy/v3/views.py

@@ -26,11 +26,14 @@
 from rest_framework.exceptions import NotFound
 from rest_framework import status
 
-from pulpcore.plugin.exceptions import DigestValidationError
-from pulpcore.plugin.models import PulpTemporaryFile, Content
+from pulpcore.plugin.models import Content
 from pulpcore.plugin.serializers import AsyncOperationResponseSerializer
-from pulpcore.plugin.viewsets import BaseFilterSet, OperationPostponedResponse
-from pulpcore.plugin.tasking import add_and_remove, dispatch
+from pulpcore.plugin.viewsets import (
+    BaseFilterSet,
+    OperationPostponedResponse,
+    SingleArtifactContentUploadViewSet,
+)
+from pulpcore.plugin.tasking import add_and_remove, dispatch, general_create
 
 from pulp_ansible.app.galaxy.v3.exceptions import ExceptionHandlerMixin
 from pulp_ansible.app.galaxy.v3.serializers import (
@@ -52,8 +55,9 @@
     DownloadLog,
 )
 from pulp_ansible.app.serializers import (
-    CollectionOneShotSerializer,
     CollectionImportDetailSerializer,
+    CollectionOneShotSerializer,
+    CollectionVersionUploadSerializer,
 )
 
 from pulp_ansible.app.galaxy.mixins import UploadGalaxyCollectionMixin
@@ -458,13 +462,15 @@ def urlpattern(*args, **kwargs):
 
 
 class CollectionUploadViewSet(
-    ExceptionHandlerMixin, viewsets.GenericViewSet, UploadGalaxyCollectionMixin
+    ExceptionHandlerMixin, UploadGalaxyCollectionMixin, SingleArtifactContentUploadViewSet
 ):
     """
     ViewSet for Collection Uploads.
     """
 
-    serializer_class = CollectionOneShotSerializer
+    queryset = None
+    endpoint_pieces = None
+    serializer_class = CollectionVersionUploadSerializer
     pulp_tag_name = "Pulp_Ansible: Artifacts Collections V3"
 
     DEFAULT_ACCESS_POLICY = _PERMISSIVE_ACCESS_POLICY
@@ -473,6 +479,16 @@ def urlpattern(*args, **kwargs):
         """Return url pattern for RBAC."""
         return "pulp_ansible/v3/collections/upload"
 
+    def _dispatch_upload_collection_task(self, args=None, kwargs=None, repository=None):
+        """
+        Dispatch an Upload Collection creation task.
+        """
+        locks = []
+        if repository:
+            locks.append(repository)
+
+        return dispatch(general_create, exclusive_resources=locks, args=args, kwargs=kwargs)
+
     @extend_schema(
         description="Create an artifact and trigger an asynchronous task to create "
         "Collection content from it.",
@@ -485,44 +501,40 @@ def create(self, request, distro_base_path):
         Dispatch a Collection creation task.
         """
         distro = get_object_or_404(AnsibleDistribution, base_path=distro_base_path)
-        serializer = self.get_serializer(data=request.data, context={"request": request})
+        repo = distro.repository
+        if repo is None:
+            if distro.repository_version is None:
+                raise serializers.ValidationError(
+                    _("Distribution must have either repository or repository_version set")
+                )
+            repo = distro.repository_version.repository
+        # Check that invalid fields were not specified
+        serializer = CollectionOneShotSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
 
-        expected_digests = {}
-        if serializer.validated_data["sha256"]:
-            expected_digests["sha256"] = serializer.validated_data["sha256"]
-        try:
-            temp_file = PulpTemporaryFile.init_and_validate(
-                serializer.validated_data["file"],
-                expected_digests=expected_digests,
-            )
-        except DigestValidationError:
-            raise serializers.ValidationError(
-                _("The provided sha256 value does not match the sha256 of the uploaded file.")
-            )
-
-        temp_file.save()
-
-        kwargs = {}
-
-        if serializer.validated_data["expected_namespace"]:
-            kwargs["expected_namespace"] = serializer.validated_data["expected_namespace"]
-
-        if serializer.validated_data["expected_name"]:
-            kwargs["expected_name"] = serializer.validated_data["expected_name"]
-
-        if serializer.validated_data["expected_version"]:
-            kwargs["expected_version"] = serializer.validated_data["expected_version"]
-
-        async_result = self._dispatch_import_collection_task(
-            temp_file.pk, distro.repository, **kwargs
+        # Check that namespace, name and version can be extracted
+        request.data["repository"] = reverse("repositories-ansible/ansible-detail", args=[repo.pk])
+        serializer = CollectionVersionUploadSerializer(
+            data=request.data, context=self.get_serializer_context()
         )
-        CollectionImport.objects.create(task_id=async_result.pk)
-
+        serializer.is_valid(raise_exception=True)
+        # Convert file to an artifact
+        task_payload = self.init_content_data(serializer, request)
+        # Dispatch create task
+        task = self._dispatch_upload_collection_task(
+            repository=serializer.validated_data["repository"],
+            args=(CollectionVersion._meta.app_label, serializer.__class__.__name__),
+            kwargs={
+                "data": task_payload,
+                "context": self.get_deferred_context(request),
+            },
+        )
+        # Create CollectionImport and response
+        CollectionImport.objects.create(task_id=task.pk)
         data = {
             "task": reverse(
                 settings.ANSIBLE_URL_NAMESPACE + "collection-imports-detail",
-                kwargs={"pk": async_result.pk},
+                kwargs={"pk": task.pk},
                 request=None,
             )
         }

pulp_ansible/app/galaxy/views.py

@@ -2,6 +2,7 @@
 
 from django.conf import settings
 from django.shortcuts import get_object_or_404, HttpResponse
+from drf_spectacular.utils import extend_schema
 from rest_framework import generics, pagination, response, views
 
 from pulpcore.plugin.models import PulpTemporaryFile
@@ -167,6 +168,7 @@ def get_queryset(self):
         """
         return Collection.objects.filter(versions__pk__in=self._distro_content).distinct()
 
+    @extend_schema(deprecated=True)
     def post(self, request, path):
         """
         Queues a task that creates a new Collection from an uploaded artifact.