Transform tests

This rewrites assertions in the pytest way, removes some fixtures in favor
of factories, and switches to requests in places where pulp_client was not
a good chioce.

pulp_ansible/tests/functional/api/collection/v3/test_collection.py

@@ -8,14 +8,12 @@
 from urllib.parse import urljoin
 
 import pytest
+import requests
 
 from pulp_smash import api, config
 from pulp_smash.pulp3.utils import gen_distribution, gen_repo
-from pulp_smash.utils import http_get
 
 from pulp_ansible.tests.functional.constants import (
-    ANSIBLE_COLLECTION_FILE_NAME,
-    ANSIBLE_COLLECTION_UPLOAD_FIXTURE_URL,
     ANSIBLE_DISTRIBUTION_PATH,
     ANSIBLE_REPO_PATH,
 )
@@ -24,6 +22,11 @@
 logger = logging.getLogger(__name__)
 
 
+class NullAuth(requests.auth.AuthBase):
+    def __call__(self, r):
+        return r
+
+
 def upload_handler(client, response):
     """Handle responses to collection upload by fetching and returning the task data."""
     response.raise_for_status()
@@ -79,9 +82,9 @@ def get_metadata_published(pulp_client, pulp_dist):
 def upload_collection(client, filename, base_path):
     """Helper to upload collections to pulp_ansible/galaxy."""
     UPLOAD_PATH = get_galaxy_url(base_path, "/v3/artifacts/collections/")
-    collection = {"file": (open(filename, "rb"))}
-
-    return client.using_handler(upload_handler).post(UPLOAD_PATH, files=collection)
+    with open(filename, "rb") as fp:
+        collection = {"file": fp}
+        return client.using_handler(upload_handler).post(UPLOAD_PATH, files=collection)
 
 
 @pytest.fixture(scope="module")
@@ -160,14 +163,6 @@ def pulp_dist(pulp_client, pulp_repo):
         pulp_client.delete(dist["pulp_href"])
 
 
-@pytest.fixture(scope="module")
-def known_collection():
-    """Fetch and prepare a known collection from Galaxy to use in an upload test."""
-    collection_content = http_get(ANSIBLE_COLLECTION_UPLOAD_FIXTURE_URL)
-    collection = {"file": (ANSIBLE_COLLECTION_FILE_NAME, collection_content)}
-    return collection
-
-
 def test_collection_upload(collection_upload):
     """Upload a new collection.
 
@@ -304,8 +299,8 @@ def test_collection_version(collection_artifact, pulp_client, collection_detail)
     assert version["name"] == collection_artifact.name
     assert version["namespace"] == {"metadata_sha256": None, "name": collection_artifact.namespace}
     assert version["version"] == "1.0.0"
-
-    tarball = open(collection_artifact.filename, "rb").read()
+    with open(collection_artifact.filename, "rb") as fp:
+        tarball = fp.read()
     assert version["artifact"]["sha256"] == hashlib.sha256(tarball).hexdigest()
     assert version["artifact"]["size"] == len(tarball)
 
@@ -331,31 +326,49 @@ def test_collection_version(collection_artifact, pulp_client, collection_detail)
     #     #             'tags': ['collectiontest']},
 
 
-def test_collection_download(collection_artifact, pulp_client, collection_detail):
+def test_collection_download(
+    bindings_cfg,
+    collection_detail,
+    collection_artifact,
+):
     """Test collection download URL.
 
     Should require authentication and redirect to a download location.
     """
-    version = pulp_client.using_handler(api.json_handler).get(
-        collection_detail["highest_version"]["href"]
+    auth = (bindings_cfg.username, bindings_cfg.password)
+
+    response = requests.get(
+        bindings_cfg.host + collection_detail["highest_version"]["href"], auth=NullAuth()
+    )
+    assert response.status_code == 401
+    response = requests.get(
+        bindings_cfg.host + collection_detail["highest_version"]["href"], auth=auth
     )
+    response.raise_for_status()
+    version = response.json()
 
     # Artifact Download Endoint
     url = version["download_url"]
 
-    tarball = open(collection_artifact.filename, "rb").read()
+    with open(collection_artifact.filename, "rb") as fp:
+        tarball = fp.read()
 
-    c = pulp_client.using_handler(api.echo_handler)
-    f = c.get(url)
-    assert f.status_code == 200, (url, f.request.headers)
-    assert f.content == tarball
+    response = requests.get(url, auth=NullAuth())
+    assert response.status_code == 401
+    response = requests.get(url, auth=auth)
+    assert response.status_code == 200, (url, response.request.headers)
+    assert response.content == tarball
 
 
-def test_collection_upload_repeat(pulp_client, collection_artifact, pulp_dist, collection_upload):
+def test_collection_upload_repeat(
+    pulp_client, ansible_collection_factory, pulp_dist, collection_upload
+):
     """
     Upload a duplicate collection.
     """
-    response = upload_collection(pulp_client, collection_artifact.filename, pulp_dist["base_path"])
+    response = upload_collection(
+        pulp_client, ansible_collection_factory().filename, pulp_dist["base_path"]
+    )
     assert response["error"] is None
     assert response["state"] == "completed"
     assert re.match(r"[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}", response["id"])

pulp_ansible/tests/functional/api/collection/v3/test_serializers.py

@@ -77,7 +77,7 @@ def test_v3_updated_at(self):
         sync_response = self.repo_api.sync(repo.pulp_href, repository_sync_data)
         monitor_task(sync_response.task)
         task = tasks.read(sync_response.task)
-        self.assertEqual(task.state, "completed")
+        assert task.state == "completed"
 
         # enumerate new data after 2nd sync ...
         collections = self.collections_api.list(distribution.base_path)
@@ -87,11 +87,11 @@ def test_v3_updated_at(self):
             "squeezer", "pulp", distribution.base_path
         ).meta.count
 
-        self.assertEqual(original_highest_version, "0.0.7")
-        self.assertEqual(highest_version, "0.0.17")
-        self.assertEqual(original_total_versions, 1)
-        self.assertEqual(total_versions, 2)
-        self.assertGreater(updated_at, original_updated_at)
+        assert original_highest_version == "0.0.7"
+        assert highest_version == "0.0.17"
+        assert original_total_versions == 1
+        assert total_versions == 2
+        assert updated_at > original_updated_at
 
     def test_v3_collection_version_from_synced_data(self):
         """Test Collection Versions V3 endpoint fields."""
@@ -111,6 +111,6 @@ def test_v3_collection_version_from_synced_data(self):
             "nxos", "cisco", distribution.base_path, "1.4.0"
         )
 
-        self.assertEqual(version.requires_ansible, ">=2.9.10,<2.11")
-        self.assertTrue("'name': 'README.md'" in str(version.files))
-        self.assertEqual(version.manifest["collection_info"]["name"], "nxos")
+        assert version.requires_ansible == ">=2.9.10,<2.11"
+        assert "'name': 'README.md'" in str(version.files)
+        assert version.manifest["collection_info"]["name"] == "nxos"

pulp_ansible/tests/functional/api/collection/v3/test_sync.py

@@ -1,14 +1,12 @@
 """Tests related to sync ansible plugin collection content type."""
 
-from pulpcore.client.pulp_ansible import (
-    AnsibleRepositorySyncURL,
-)
-from pulp_smash.pulp3.bindings import PulpTaskError
+from pulpcore.client.pulp_ansible import AnsibleRepositorySyncURL
+
+from pulpcore.tests.functional.utils import PulpTaskError
 
 from pulp_ansible.tests.functional.utils import (
     gen_ansible_remote,
     monitor_task,
-    tasks,
 )
 from pulp_ansible.tests.functional.utils import TestCaseUsingBindings
 
@@ -54,9 +52,9 @@ def test_sync_collections_from_pulp(self):
         first_content = self.cv_api.list(
             repository_version=f"{self.first_repo.pulp_href}versions/1/"
         )
-        self.assertGreaterEqual(len(first_content.results), 1)
+        assert len(first_content.results) >= 1
         second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/")
-        self.assertGreaterEqual(len(second_content.results), 1)
+        assert len(second_content.results) >= 1
 
     def test_sync_collections_from_pulp_using_mirror_second_time(self):
         """Test sync collections from pulp server using a mirror option the second time."""
@@ -78,9 +76,9 @@ def test_sync_collections_from_pulp_using_mirror_second_time(self):
         second_repo = self._create_repo_and_sync_with_remote(second_remote)
 
         first_content = self.cv_api.list(repository_version=f"{first_repo.pulp_href}versions/1/")
-        self.assertGreaterEqual(len(first_content.results), 1)
+        assert len(first_content.results) >= 1
         second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/")
-        self.assertGreaterEqual(len(second_content.results), 1)
+        assert len(second_content.results) >= 1
 
     def test_sync_collection_named_api(self):
         """Test sync collections from pulp server."""
@@ -97,8 +95,8 @@ def test_sync_collection_named_api(self):
 
         collection = self.collections_v3api.read("api", "rswaf", distribution.base_path)
 
-        self.assertEqual("api", collection.name)
-        self.assertEqual("rswaf", collection.namespace)
+        assert "api" == collection.name
+        assert "rswaf" == collection.namespace
 
     def test_noop_resync_collections_from_pulp(self):
         """Test whether sync yields no-op when repo hasn't changed since last sync."""
@@ -114,20 +112,19 @@ def test_noop_resync_collections_from_pulp(self):
         second_repo = self._create_repo_with_attached_remote_and_sync(second_remote)
 
         second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/")
-        self.assertGreaterEqual(len(second_content.results), 1)
+        assert len(second_content.results) >= 1
 
         # Resync
         repository_sync_data = AnsibleRepositorySyncURL(
             remote=second_remote.pulp_href, optimize=True
         )
         sync_response = self.repo_api.sync(second_repo.pulp_href, repository_sync_data)
-        monitor_task(sync_response.task)
+        task = monitor_task(sync_response.task)
         second_repo = self.repo_api.read(second_repo.pulp_href)
-        task = tasks.read(sync_response.task)
 
         msg = "no-op: {url} did not change since last sync".format(url=second_remote.url)
         messages = [r.message for r in task.progress_reports]
-        self.assertIn(msg, str(messages))
+        assert msg in str(messages)
 
     def test_noop_resync_with_mirror_from_pulp(self):
         """Test whether no-op sync with mirror=True doesn't remove repository content."""
@@ -143,21 +140,20 @@ def test_noop_resync_with_mirror_from_pulp(self):
         second_repo = self._create_repo_with_attached_remote_and_sync(second_remote)
 
         second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/")
-        self.assertGreaterEqual(len(second_content.results), 1)
+        assert len(second_content.results) >= 1
 
         # Resync
         repository_sync_data = AnsibleRepositorySyncURL(
             remote=second_remote.pulp_href, optimize=True, mirror=True
         )
         sync_response = self.repo_api.sync(second_repo.pulp_href, repository_sync_data)
-        monitor_task(sync_response.task)
+        task = monitor_task(sync_response.task)
         second_repo = self.repo_api.read(second_repo.pulp_href)
-        self.assertEqual(int(second_repo.latest_version_href[-2]), 1)
-        task = tasks.read(sync_response.task)
+        assert int(second_repo.latest_version_href[-2]) == 1
 
         msg = "no-op: {url} did not change since last sync".format(url=second_remote.url)
         messages = [r.message for r in task.progress_reports]
-        self.assertIn(msg, str(messages))
+        assert msg in str(messages)
 
     def test_update_requirements_file(self):
         """Test requirements_file update."""
@@ -171,15 +167,15 @@ def test_update_requirements_file(self):
         self.addCleanup(self.remote_collection_api.delete, remote.pulp_href)
 
         repo = self._create_repo_with_attached_remote_and_sync(remote)
-        self.assertIsNotNone(repo.last_synced_metadata_time)
+        assert repo.last_synced_metadata_time is not None
 
         response = self.remote_collection_api.partial_update(
             remote.pulp_href, {"requirements_file": "collections:\n  - ansible.posix"}
         )
         monitor_task(response.task)
 
         repo = self.repo_api.read(repo.pulp_href)
-        self.assertIsNone(repo.last_synced_metadata_time)
+        assert repo.last_synced_metadata_time is None
 
     def test_sync_with_missing_collection(self):
         """Test that syncing with a non-present collection gives a useful error."""
@@ -197,4 +193,4 @@ def test_sync_with_missing_collection(self):
 
         task_result = cm.exception.task.to_dict()
         msg = "absent.not_present does not exist"
-        self.assertIn(msg, task_result["error"]["description"], task_result["error"]["description"])
+        assert msg in task_result["error"]["description"], task_result["error"]["description"]

pulp_ansible/tests/functional/utils.py

@@ -19,7 +19,6 @@
 from pulpcore.client.pulpcore import (
     ApiClient as CoreApiClient,
     TasksApi,
-    StatusApi,
 )
 from pulpcore.client.pulp_ansible import (
     ApiClient as AnsibleApiClient,