Add upload test case for collection

Add upload test case for collection.

closes: #5262

CHANGES/5262.misc

@@ -0,0 +1 @@
+Add upload test case

pulp_ansible/tests/functional/api/collection/test_upload.py

@@ -0,0 +1,55 @@
+# coding=utf-8
+"""Tests related to upload of collections."""
+import hashlib
+import unittest
+from urllib.parse import urljoin
+
+from pulp_smash import api, config
+from pulp_smash.pulp3.utils import delete_orphans
+from pulp_smash.utils import http_get
+from requests.exceptions import HTTPError
+
+from pulp_ansible.tests.functional.constants import (
+    ANSIBLE_COLLECTION_UPLOAD_FIXTURE_URL,
+    COLLECTION_METADATA,
+)
+from pulp_ansible.tests.functional.utils import set_up_module as setUpModule  # noqa:F401
+
+
+class UploadCollectionTestCase(unittest.TestCase):
+    """Upload a collection."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create class-wide variables."""
+        cls.cfg = config.get_config()
+        delete_orphans(cls.cfg)
+        cls.client = api.Client(cls.cfg)
+
+        cls.collection = {"file": http_get(ANSIBLE_COLLECTION_UPLOAD_FIXTURE_URL)}
+        cls.collection_sha256 = hashlib.sha256(cls.collection["file"]).hexdigest()
+
+    def test_collection_upload(self):
+        """Upload a collection.
+
+        This test targets the following issue:
+
+        * `Pulp #5262 <https://pulp.plan.io/issues/5262>`_
+        """
+        UPLOAD_PATH = urljoin(self.cfg.get_base_url(), "ansible/collections/")
+        response = self.client.post(UPLOAD_PATH, files=self.collection)
+
+        for key, value in response.items():
+            with self.subTest(key=key):
+                if key in COLLECTION_METADATA.keys():
+                    self.assertEqual(COLLECTION_METADATA[key], value, response)
+
+        self.assertEqual(response["sha256"], self.collection_sha256, response)
+
+        with self.assertRaises(HTTPError) as ctx:
+            self.client.using_handler(api.code_handler).post(UPLOAD_PATH, files=self.collection)
+
+        for key in ("artifact", "already", "exists"):
+            self.assertIn(key, ctx.exception.response.json()[0].lower(), ctx.exception.response)
+
+        delete_orphans(self.cfg)

pulp_ansible/tests/functional/constants.py

@@ -7,6 +7,8 @@
     CONTENT_PATH,
 )
 
+GALAXY_ANSIBLE_BASE_URL = "https://galaxy.ansible.com"
+
 ANSIBLE_ROLE_NAME = "ansible.role"
 
 ANSIBLE_ROLE_CONTENT_PATH = urljoin(CONTENT_PATH, "ansible/roles/")
@@ -17,7 +19,7 @@
 
 ANSIBLE_PUBLISHER_PATH = urljoin(BASE_PUBLISHER_PATH, "ansible/ansible/")
 
-ANSIBLE_GALAXY_URL = "https://galaxy.ansible.com/api/v1/roles/"
+ANSIBLE_GALAXY_URL = urljoin(GALAXY_ANSIBLE_BASE_URL, "api/v1/roles/")
 
 NAMESPACE_ANSIBLE = "?namespace__name=ansible"
 
@@ -33,14 +35,14 @@
 
 ANSIBLE_ELASTIC_FIXTURE_URL = urljoin(ANSIBLE_GALAXY_URL, NAMESPACE_ELASTIC)
 
-ANSIBLE_FIXTURE_CONTENT_SUMMARY = {ANSIBLE_ROLE_NAME: 5}
-
 ANSIBLE_FIXTURE_COUNT = 5
 
+ANSIBLE_FIXTURE_CONTENT_SUMMARY = {ANSIBLE_ROLE_NAME: ANSIBLE_FIXTURE_COUNT}
+
 # FIXME: replace this with the location of one specific content unit of your choosing
 ANSIBLE_URL = urljoin(ANSIBLE_FIXTURE_URL, "")
 
-ANSIBLE_GALAXY_COLLECTION_URL = "https://galaxy.ansible.com/api/v2/collections/"
+ANSIBLE_GALAXY_COLLECTION_URL = urljoin(GALAXY_ANSIBLE_BASE_URL, "api/v2/collections/")
 
 ANSIBLE_COLLECTION_FIXTURE_URL = urljoin(ANSIBLE_GALAXY_COLLECTION_URL, NAMESPACE_TESTING)
 
@@ -55,3 +57,11 @@
 ANSIBLE_COLLECTION_FIXTURE_SUMMARY = {
     ANSIBLE_COLLECTION_CONTENT_NAME: ANSIBLE_COLLECTION_FIXTURE_COUNT
 }
+
+COLLECTION_METADATA = {"name": "k8s_demo_collection", "version": "0.0.3"}
+"""Metadata was extracted from
+https://galaxy.ansible.com/api/v2/collections/testing/k8s_demo_collection/versions/0.0.3/"""
+
+ANSIBLE_COLLECTION_UPLOAD_FIXTURE_URL = urljoin(
+    GALAXY_ANSIBLE_BASE_URL, "download/testing-k8s_demo_collection-0.0.3.tar.gz"
+)