Use pulp_ansible collection upload refactor

[noissue]
Required PR: pulp/pulp_ansible#1176

galaxy_ng/app/api/v3/viewsets/collection.py

@@ -46,22 +46,23 @@ class CollectionUploadViewSet(api_base.LocalSettingsMixin,
     parser_classes = [AnsibleGalaxy29MultiPartParser]
     serializer_class = CollectionUploadSerializer
 
-    def _dispatch_import_collection_task(self, temp_file_pk, repository=None, **kwargs):
-        """Dispatch a pulp task started on upload of collection version."""
-        locks = []
-        context = super().get_serializer_context()
-        request = context.get("request", None)
-
-        kwargs["temp_file_pk"] = temp_file_pk
-        kwargs["username"] = request.user.username
-
-        if repository:
-            locks.append(repository)
-            kwargs["repository_pk"] = repository.pk
+    def _get_upload_task_and_kwargs(self, request):
+        """Temporary hook to change the upload task until viewset is moved to pulp_ansible."""
+        upload_task, task_kwargs = super()._get_upload_task_and_kwargs(request)
+        # Set galaxy-specific kwargs
+        galaxy_kwargs = {}
+        exclusive_resources = task_kwargs.pop("exclusive_resources")
+        galaxy_kwargs["exclusive_resources"] = exclusive_resources
+        # import_and_* signatures: (inbound_repository_pk, username)
+        galaxy_kwargs["args"] = (exclusive_resources[0].pk, request.user.username)
+        galaxy_kwargs["kwargs"]["general_create_kwargs"] = task_kwargs
+        for field in ("expected_name", "expected_namespace", "expected_version"):
+            # Serializer data has expected_namespace/name/version
+            galaxy_kwargs["kwargs"][field] = task_kwargs["kwargs"]["data"][field]
 
         if settings.GALAXY_REQUIRE_CONTENT_APPROVAL:
-            return dispatch(import_and_move_to_staging, exclusive_resources=locks, kwargs=kwargs)
-        return dispatch(import_and_auto_approve, exclusive_resources=locks, kwargs=kwargs)
+            return import_and_move_to_staging, galaxy_kwargs
+        return import_and_auto_approve, galaxy_kwargs
 
     # Wrap super().create() so we can create a galaxy_ng.app.models.CollectionImport based on the
     # the import task and the collection artifact details

galaxy_ng/app/tasks/publishing.py

@@ -4,7 +4,7 @@
 from django.contrib.contenttypes.models import ContentType
 from django.utils.translation import gettext_lazy as _
 from pulp_ansible.app.models import AnsibleDistribution, AnsibleRepository, CollectionVersion
-from pulp_ansible.app.tasks.collections import import_collection
+from pulpcore.plugin.tasking import general_create
 from pulpcore.plugin.models import Task
 from pulpcore.plugin.models import SigningService
 
@@ -34,22 +34,16 @@ def get_created_collection_versions():
     return created_collection_versions
 
 
-def import_and_move_to_staging(temp_file_pk, **kwargs):
+def import_and_move_to_staging(inbound_repository_pk, username, **kwargs):
     """Import collection version and move to staging repository.
 
-    Custom task to call pulp_ansible's import_collection() task then
+    Custom task to call pulpcore's general_create() task then
     enqueue two tasks to add to staging repo and remove from inbound repo.
 
     This task will not wait for the enqueued tasks to finish.
     """
-    inbound_repository_pk = kwargs.get('repository_pk')
-    import_collection(
-        temp_file_pk=temp_file_pk,
-        repository_pk=inbound_repository_pk,
-        expected_namespace=kwargs['expected_namespace'],
-        expected_name=kwargs['expected_name'],
-        expected_version=kwargs['expected_version'],
-    )
+    general_kwargs = kwargs.pop("general_create_kwargs")
+    general_create(**general_kwargs)
 
     try:
         staging_repo = AnsibleDistribution.objects.get(name=STAGING_NAME).repository
@@ -65,28 +59,22 @@ def import_and_move_to_staging(temp_file_pk, **kwargs):
 
         if settings.GALAXY_ENABLE_API_ACCESS_LOG:
             _log_collection_upload(
-                kwargs["username"],
+                username,
                 kwargs["expected_namespace"],
                 kwargs["expected_name"],
                 kwargs["expected_version"]
             )
 
 
-def import_and_auto_approve(temp_file_pk, **kwargs):
+def import_and_auto_approve(inbound_repository_pk, username, **kwargs):
     """Import collection version and automatically approve.
 
-    Custom task to call pulp_ansible's import_collection() task
+    Custom task to call pulpcore's general_create() task
     then automatically approve collection version so no
     manual approval action needs to occur.
     """
-    inbound_repository_pk = kwargs.get('repository_pk')
-    import_collection(
-        temp_file_pk=temp_file_pk,
-        repository_pk=inbound_repository_pk,
-        expected_namespace=kwargs['expected_namespace'],
-        expected_name=kwargs['expected_name'],
-        expected_version=kwargs['expected_version'],
-    )
+    general_kwargs = kwargs.pop("general_create_kwargs")
+    general_create(**general_kwargs)
 
     try:
         golden_repo = AnsibleDistribution.objects.get(name=GOLDEN_NAME).repository

galaxy_ng/tests/unit/app/test_tasks.py

@@ -80,25 +80,25 @@ def test_task_move_content(self):
         self.assertEqual(repo2_version_number + 1, repo2.latest_version().number)
 
     @mock.patch('galaxy_ng.app.tasks.publishing.get_created_collection_versions')
-    @mock.patch('galaxy_ng.app.tasks.publishing.import_collection')
+    @mock.patch('galaxy_ng.app.tasks.publishing.general_create')
     @mock.patch('galaxy_ng.app.tasks.promotion.dispatch')
-    def test_import_and_auto_approve(self, mocked_dispatch, mocked_import, mocked_get_created):
+    def test_import_and_auto_approve(self, mocked_dispatch, mocked_create, mocked_get_created):
         inbound_repo = AnsibleRepository.objects.get(name=staging_name)
 
         golden_repo = AnsibleRepository.objects.get(name=golden_name)
 
         mocked_get_created.return_value = [self.collection_version]
 
         import_and_auto_approve(
-            self.pulp_temp_file.pk,
-            repository_pk=inbound_repo.pk,
+            inbound_repo.pk,
+            '',  # username
             expected_namespace='',
             expected_name='',
             expected_version='',
-            username='',
+            general_create_kwargs={},
         )
 
-        self.assertTrue(mocked_import.call_count == 1)
+        self.assertTrue(mocked_create.call_count == 1)
         self.assertTrue(mocked_dispatch.call_count == 1)
 
         # test cannot find golden repo
@@ -107,18 +107,18 @@ def test_import_and_auto_approve(self, mocked_dispatch, mocked_import, mocked_ge
         mocked_get_created.side_effect = AnsibleDistribution.DoesNotExist
         with self.assertRaises(AnsibleDistribution.DoesNotExist):
             import_and_auto_approve(
-                self.artifact.pk,
-                repository_pk=inbound_repo.pk,
+                inbound_repo.pk,
+                '',  # username
                 expected_namespace='',
                 expected_name='',
                 expected_version='',
-                username='',
+                general_create_kwargs={},
             )
 
     @mock.patch('galaxy_ng.app.tasks.publishing.get_created_collection_versions')
-    @mock.patch('galaxy_ng.app.tasks.publishing.import_collection')
+    @mock.patch('galaxy_ng.app.tasks.publishing.general_create')
     @mock.patch('galaxy_ng.app.tasks.promotion.dispatch')
-    def test_import_and_move_to_staging(self, mocked_dispatch, mocked_import, mocked_get_created):
+    def test_import_and_move_to_staging(self, mocked_dispatch, mocked_create, mocked_get_created):
         staging_repo = AnsibleRepository.objects.get(name=staging_name)
 
         inbound_name = 'the_incoming_repo'
@@ -130,15 +130,15 @@ def test_import_and_move_to_staging(self, mocked_dispatch, mocked_import, mocked
         mocked_get_created.return_value = [self.collection_version]
 
         import_and_move_to_staging(
-            self.pulp_temp_file.pk,
-            repository_pk=inbound_repo.pk,
+            inbound_repo.pk,
+            '',  # username
             expected_namespace='',
             expected_name='',
             expected_version='',
-            username='',
+            general_create_kwargs={},
         )
 
-        self.assertTrue(mocked_import.call_count == 1)
+        self.assertTrue(mocked_create.call_count == 1)
         self.assertTrue(mocked_dispatch.call_count == 1)
 
         # test cannot find staging repo
@@ -147,12 +147,12 @@ def test_import_and_move_to_staging(self, mocked_dispatch, mocked_import, mocked
         mocked_get_created.side_effect = AnsibleDistribution.DoesNotExist
         with self.assertRaises(AnsibleDistribution.DoesNotExist):
             import_and_move_to_staging(
-                self.pulp_temp_file.pk,
-                repository_pk=inbound_repo.pk,
+                inbound_repo.pk,
+                '',  # username
                 expected_namespace='',
                 expected_name='',
                 expected_version='',
-                username='',
+                general_create_kwargs={},
             )
 
     def test_log_collection_upload(self):