Merge pull request #387 from gerrod3/twine-upload-fix

Simplified twine upload to create one task per distribution uploaded

Author: gerrod3

pulp_python/app/pypi/views.py

@@ -124,18 +124,25 @@ def upload(self, request, path):
         if repo_content.filter(filename=filename).exists():
             return HttpResponseBadRequest(reason=f"Package {filename} already exists in index")
 
-        return self.upload_package(repo, artifact, filename, request.session)
+        if settings.PYTHON_GROUP_UPLOADS:
+            return self.upload_package_group(repo, artifact, filename, request.session)
 
-    def upload_package(self, repo, artifact, filename, session):
+        result = dispatch(tasks.upload, [artifact, repo],
+                          kwargs={"artifact_sha256": artifact.sha256,
+                                  "filename": filename,
+                                  "repository_pk": str(repo.pk)})
+        return Response(data={"task": reverse('tasks-detail', args=[result.pk], request=None)})
+
+    def upload_package_group(self, repo, artifact, filename, session):
         """Steps 4 & 5, spawns tasks to add packages to index."""
         start_time = datetime.now(tz=timezone.utc) + timedelta(seconds=5)
         task = "updated"
         if not session.get("start"):
-            task = self.create_upload_task(session, repo, artifact, filename, start_time)
+            task = self.create_group_upload_task(session, repo, artifact, filename, start_time)
         else:
             sq = Session.objects.select_for_update(nowait=True).filter(pk=session.session_key)
-            with transaction.atomic():
-                try:
+            try:
+                with transaction.atomic():
                     sq.first()
                     try:
                         current_start = datetime.fromisoformat(session['start'])
@@ -149,19 +156,19 @@ def upload_package(self, repo, artifact, filename, session):
                         session.save()
                     else:
                         raise DatabaseError
-                except DatabaseError:
-                    session.cycle_key()
-                    task = self.create_upload_task(session, repo, artifact, filename, start_time)
+            except DatabaseError:
+                session.cycle_key()
+                task = self.create_group_upload_task(session, repo, artifact, filename, start_time)
         data = {"session": session.session_key, "task": task, "task_start_time": start_time}
         return Response(data=data)
 
-    def create_upload_task(self, cur_session, repository, artifact, filename, start_time):
+    def create_group_upload_task(self, cur_session, repository, artifact, filename, start_time):
         """Creates the actual task that adds the packages to the index."""
         cur_session['start'] = str(start_time)
         cur_session['artifacts'] = [(str(artifact.sha256), filename)]
         cur_session.modified = False
         cur_session.save()
-        result = dispatch(tasks.upload, [artifact, repository],
+        result = dispatch(tasks.upload_group, [artifact, repository],
                           kwargs={"session_pk": str(cur_session.session_key),
                                   "repository_pk": str(repository.pk)})
         return reverse('tasks-detail', args=[result.pk], request=None)

pulp_python/app/settings.py

@@ -0,0 +1 @@
+PYTHON_GROUP_UPLOADS = False

pulp_python/app/tasks/__init__.py

@@ -4,4 +4,4 @@
 
 from .publish import publish  # noqa:F401
 from .sync import sync  # noqa:F401
-from .upload import upload  # noqa:F401
+from .upload import upload, upload_group  # noqa:F401

pulp_python/app/tasks/upload.py

@@ -30,7 +30,24 @@
 }
 
 
-def upload(session_pk, repository_pk=None):
+def upload(artifact_sha256, filename, repository_pk=None):
+    """
+    Uploads a Python Package to Pulp
+
+    Args:
+        artifact_sha256: the sha256 of the artifact in Pulp to create a package from
+        filename: the full filename of the package to create
+        repository_pk: the optional pk of the repository to add the content to
+    """
+    pre_check = PythonPackageContent.objects.filter(sha256=artifact_sha256)
+    content_to_add = pre_check or create_content(artifact_sha256, filename)
+    if repository_pk:
+        repository = PythonRepository.objects.get(pk=repository_pk)
+        with repository.new_version() as new_version:
+            new_version.add_content(content_to_add)
+
+
+def upload_group(session_pk, repository_pk=None):
     """
     Uploads a Python Package to Pulp
 

pulp_python/tests/functional/api/test_pypi_apis.py

@@ -123,31 +123,7 @@ def test_package_upload(self):
         )
 
     def test_package_upload_session(self):
-        """Tests that multiple packages can be uploaded in one session."""
-        repo, distro = self._create_empty_repo_and_distribution()
-        url = urljoin(PYPI_HOST, distro.base_path + "/legacy/")
-        session = requests.Session()
-        response = session.post(
-            url,
-            data={"sha256_digest": PYTHON_EGG_SHA256},
-            files={"content": open(self.egg, "rb")},
-        )
-        response2 = session.post(
-            url,
-            data={"sha256_digest": PYTHON_WHEEL_SHA256},
-            files={"content": open(self.wheel, "rb")},
-        )
-        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response2.status_code, 200)
-        response, response2 = response.json(), response2.json()
-        self.assertEqual(response["session"], response2["session"])
-        self.assertEqual(response2["task"], "updated")
-        monitor_task(response["task"])
-        content = get_added_content_summary(repo, f"{repo.versions_href}1/")
-        self.assertDictEqual({PYTHON_CONTENT_NAME: 2}, content)
-
-    def test_package_upload_session_long(self):
-        """Tests that long uploads will be broken up into multiple tasks."""
+        """Tests that multiple uploads will be broken up into multiple tasks."""
         repo, distro = self._create_empty_repo_and_distribution()
         url = urljoin(PYPI_HOST, distro.base_path + "/legacy/")
         session = requests.Session()
@@ -166,8 +142,6 @@ def test_package_upload_session_long(self):
         )
         self.assertEqual(response2.status_code, 200)
         response2 = response2.json()
-        self.assertNotEqual("updated", response2["task"])
-        self.assertNotEqual(response["session"], response2["session"])
         self.assertNotEqual(response["task"], response2["task"])
         monitor_task(response2["task"])
         content = get_content_summary(repo, f"{repo.versions_href}2/")
@@ -208,9 +182,11 @@ def test_twine_upload(self):
             capture_output=True,
             check=True,
         )
-        task = task_api.list(reserved_resources_record=repo.pulp_href).results[0]
-        monitor_task(task.pulp_href)
-        content = get_added_content_summary(repo, f"{repo.versions_href}1/")
+        tasks = task_api.list(reserved_resources_record=[repo.pulp_href]).results
+        for task in reversed(tasks):
+            t = monitor_task(task.pulp_href)
+            repo_ver_href = t.created_resources[-1]
+        content = get_content_summary(repo, f"{repo_ver_href}")
         self.assertDictEqual({PYTHON_CONTENT_NAME: 2}, content)
 
         # Test re-uploading same packages gives error